From 5e7b83085c2461af0c5ead221474fdc5195d6989 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Fri, 16 Dec 2022 14:22:42 -0500 Subject: [PATCH 1/3] Fix display of cost/dynamic cost. Fixes #929 --- ziti/cmd/fabric/list.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ziti/cmd/fabric/list.go b/ziti/cmd/fabric/list.go index d41e9ad70..8e7ca63e7 100644 --- a/ziti/cmd/fabric/list.go +++ b/ziti/cmd/fabric/list.go @@ -241,9 +241,9 @@ func outputTerminators(o *api.Options, children []*gabs.Container, pagingInfo *a binding := api.GetJsonString(entity, "binding") address := api.GetJsonString(entity, "address") instanceId := api.GetJsonString(entity, "instanceId") - staticCost := api.GetJsonString(entity, "cost") + staticCost := api.GetJsonValue(entity, "cost") precedence := api.GetJsonString(entity, "precedence") - dynamicCost := api.GetJsonString(entity, "dynamicCost") + dynamicCost := api.GetJsonValue(entity, "dynamicCost") hostId := api.GetJsonString(entity, "hostId") t.AppendRow(table.Row{id, service, router, binding, address, instanceId, staticCost, precedence, dynamicCost, hostId}) From 5209a70534b1aca924caad8ffbe8d94a3277b22d Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Fri, 16 Dec 2022 14:23:02 -0500 Subject: [PATCH 2/3] Don't validate router unless router is being updated. Fixes #928 --- ziti/cmd/fabric/update_terminator.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/ziti/cmd/fabric/update_terminator.go b/ziti/cmd/fabric/update_terminator.go index 9541f8d15..65defb261 100644 --- a/ziti/cmd/fabric/update_terminator.go +++ b/ziti/cmd/fabric/update_terminator.go @@ -75,13 +75,13 @@ func newUpdateTerminatorCmd(p common.OptionsProvider) *cobra.Command { func runUpdateTerminator(o *updateTerminatorOptions) (err error) { entityData := gabs.New() - router, err := api.MapNameToID(util.FabricAPI, "routers", &o.Options, o.router) - if err != nil { - return err - } - change := false if o.Cmd.Flags().Changed("router") { + router, err := api.MapNameToID(util.FabricAPI, "routers", &o.Options, o.router) + if err != nil { + return err + } + api.SetJSONValue(entityData, router, "router") change = true } From 7701c031ad1310d140071b3e4d60a5e2165993e4 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 20 Dec 2022 14:34:36 -0500 Subject: [PATCH 3/3] Update deps and changelog. Add stream events. Remove stream circuits/metrics --- CHANGELOG.md | 30 ++++ go.mod | 12 +- go.sum | 24 +-- ziti/cmd/api/ws.go | 2 - ziti/cmd/fabric/root.go | 3 +- ziti/cmd/fabric/stream_circuits.go | 94 ------------ ziti/cmd/fabric/stream_events.go | 232 +++++++++++++++++++++++++++++ ziti/cmd/fabric/stream_metrics.go | 135 ----------------- ziti/router/run.go | 2 +- 9 files changed, 282 insertions(+), 252 deletions(-) delete mode 100644 ziti/cmd/fabric/stream_circuits.go create mode 100644 ziti/cmd/fabric/stream_events.go delete mode 100644 ziti/cmd/fabric/stream_metrics.go diff --git a/CHANGELOG.md b/CHANGELOG.md index af1e6b3bb..7ab25c7ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,33 @@ +# Release 0.27.1 + +## What's New + +* Event streaming over websocket + * `ziti fabric stream events` + * Events use same JSON formatting as the file based streaming + * Plain Text formatting removed + * Individual streaming of metrics/circuits removed in favor of unified events streaming +* Improvements to router/tunneler terminator creation + * Create terminator requests are now idempotent, so repeated requests will not result in multiple terminators + * Create terminator requests are now asynchronous, so responses will no longer get timed out + * There is new timer metric from routers, timing how long terminator creates take: `xgress_edge_tunnel.terminator.create_timer` + +## Component Updates and Bug Fixes + +* github.com/openziti/edge: [v0.24.75 -> v0.24.86](https://github.com/openziti/edge/compare/v0.24.75...v0.24.86) + * [Issue #1272](https://github.com/openziti/edge/issues/1272) - Mark xgress_edge and xgress_edge_tunnel created terminators as system entity + * [Issue #1270](https://github.com/openziti/edge/issues/1270) - Make xgress_edge_tunnel service hosting more scalabe + * [Issue #1268](https://github.com/openziti/edge/issues/1268) - session deletion can get stalled by restarts + +* github.com/openziti/fabric: [v0.21.36 -> v0.22.1](https://github.com/openziti/fabric/compare/v0.21.36...v0.22.1) + * [Issue #563](https://github.com/openziti/fabric/issues/563) - Allow streaming events over webscocket, replacing stream circuits and stream metrics + * [Issue #552](https://github.com/openziti/fabric/issues/552) - Add minimum cost delta for smart routing + * [Issue #558](https://github.com/openziti/fabric/issues/558) - Allow terminators to be marked as system entities + +* github.com/openziti/ziti: [v0.27.0 -> v0.27.1](https://github.com/openziti/ziti/compare/v0.27.0...v0.27.1) + * [Issue #928](https://github.com/openziti/ziti/issues/928) - ziti fabric update terminator should not require setting router + * [Issue #929](https://github.com/openziti/ziti/issues/929) - zit fabric list terminators isn't showing cost or dynamic cost + # Release 0.27.0 ## What's New diff --git a/go.mod b/go.mod index 11e5620ac..4a93d321d 100644 --- a/go.mod +++ b/go.mod @@ -22,8 +22,8 @@ require ( github.com/michaelquigley/pfxlog v0.6.9 github.com/openziti/agent v1.0.7 github.com/openziti/channel/v2 v2.0.24 - github.com/openziti/edge v0.24.75 - github.com/openziti/fabric v0.21.36 + github.com/openziti/edge v0.24.86 + github.com/openziti/fabric v0.22.1 github.com/openziti/foundation/v2 v2.0.9 github.com/openziti/identity v1.0.28 github.com/openziti/runzmd v1.0.7 @@ -100,7 +100,7 @@ require ( github.com/google/uuid v1.3.0 // indirect github.com/gorilla/handlers v1.5.1 // indirect github.com/gorilla/mux v1.8.0 // indirect - github.com/hashicorp/go-hclog v1.3.1 // indirect + github.com/hashicorp/go-hclog v1.4.0 // indirect github.com/hashicorp/go-immutable-radix v1.3.1 // indirect github.com/hashicorp/go-msgpack v0.5.5 // indirect github.com/hashicorp/golang-lru v0.6.0 // indirect @@ -126,7 +126,7 @@ require ( github.com/mattn/go-isatty v0.0.16 // indirect github.com/mattn/go-runewidth v0.0.14 // indirect github.com/mattn/go-tty v0.0.3 // indirect - github.com/mdlayher/netlink v1.7.0 // indirect + github.com/mdlayher/netlink v1.7.1 // indirect github.com/mdlayher/socket v0.4.0 // indirect github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d // indirect github.com/miekg/dns v1.1.50 // indirect @@ -150,7 +150,7 @@ require ( github.com/pelletier/go-toml v1.9.4 // indirect github.com/pkg/term v1.2.0-beta.2 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect - github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c // indirect + github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b // indirect github.com/rivo/uniseg v0.4.3 // indirect github.com/rodaine/table v1.0.1 // indirect github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e // indirect @@ -173,7 +173,7 @@ require ( go.opentelemetry.io/otel v1.11.1 // indirect go.opentelemetry.io/otel/trace v1.11.1 // indirect golang.org/x/crypto v0.4.0 // indirect - golang.org/x/exp v0.0.0-20221211140036-ad323defaf05 // indirect + golang.org/x/exp v0.0.0-20221217163422-3c43f8badb15 // indirect golang.org/x/image v0.2.0 // indirect golang.org/x/mod v0.6.0 // indirect golang.org/x/sync v0.1.0 // indirect diff --git a/go.sum b/go.sum index 81bceaca0..fbf0ec195 100644 --- a/go.sum +++ b/go.sum @@ -447,8 +447,8 @@ github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtng github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80= github.com/hashicorp/go-hclog v0.9.1/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ= github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ= -github.com/hashicorp/go-hclog v1.3.1 h1:vDwF1DFNZhntP4DAjuTpOw3uEgMUpXh1pB5fW9DqHpo= -github.com/hashicorp/go-hclog v1.3.1/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= +github.com/hashicorp/go-hclog v1.4.0 h1:ctuWFGrhFha8BnnzxqeRGidlEcQkDyL5u8J8t5eA11I= +github.com/hashicorp/go-hclog v1.4.0/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc= github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= @@ -612,8 +612,8 @@ github.com/mattn/go-tty v0.0.0-20180219170247-931426f7535a/go.mod h1:XPvLUNfbS4f github.com/mattn/go-tty v0.0.3 h1:5OfyWorkyO7xP52Mq7tB36ajHDG5OHrmBGIS/DtakQI= github.com/mattn/go-tty v0.0.3/go.mod h1:ihxohKRERHTVzN+aSVRwACLCeqIoZAWpoICkkvrWyR0= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= -github.com/mdlayher/netlink v1.7.0 h1:ZNGI4V7i1fJ94DPYtWhI/R85i/Q7ZxnuhUJQcJMoodI= -github.com/mdlayher/netlink v1.7.0/go.mod h1:nKO5CSjE/DJjVhk/TNp6vCE1ktVxEA8VEh8drhZzxsQ= +github.com/mdlayher/netlink v1.7.1 h1:FdUaT/e33HjEXagwELR8R3/KL1Fq5x3G5jgHLp/BTmg= +github.com/mdlayher/netlink v1.7.1/go.mod h1:nKO5CSjE/DJjVhk/TNp6vCE1ktVxEA8VEh8drhZzxsQ= github.com/mdlayher/socket v0.4.0 h1:280wsy40IC9M9q1uPGcLBwXpcTQDtoGwVt+BNoITxIw= github.com/mdlayher/socket v0.4.0/go.mod h1:xxFqz5GRCUN3UEOm9CZqEJsAbe1C8OwSK46NlmWuVoc= github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE= @@ -698,10 +698,10 @@ github.com/openziti/channel/v2 v2.0.24 h1:mg9/+VMyXKXfbTZ06quwjRcTQAqU/MEMm+KxAb github.com/openziti/channel/v2 v2.0.24/go.mod h1:Y75U/q7l2U+Iib+rfSXO0IscYvksfwWjgdGDXcM/WwY= 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.24.75 h1:siioCtoR1ZgiLO1NRqyUT+cEO9UtN6jNTKzs/978ZOk= -github.com/openziti/edge v0.24.75/go.mod h1:UYTO4JPv8Wud7rWbrA5XImyGUZIMdliSTRCXe3ZYXN0= -github.com/openziti/fabric v0.21.36 h1:AIqNFL+W0uPhItroVcr7+9uJB4fd9S46y0c9I/7P2bk= -github.com/openziti/fabric v0.21.36/go.mod h1:HBtR0TQ7kLWjp08ynloTV2CaJNRdFbybsEUlJQ0xjwo= +github.com/openziti/edge v0.24.86 h1:41TxCaFridKp9T5LDwiULsfIUDVrB7SMyykvEzriTGk= +github.com/openziti/edge v0.24.86/go.mod h1:qAF4uLkcmX8Y+t3yzuocq10Wz2D6rXkRRYLGO5AgRd4= +github.com/openziti/fabric v0.22.1 h1:kwHib2kfoVXRExhA4RhmybByE42ov1Lu/P86ZJ0GufE= +github.com/openziti/fabric v0.22.1/go.mod h1:i17oyfaF24Kenvh5nEq0qkBIL4AUAoh08D6WQW1dVbE= github.com/openziti/foundation/v2 v2.0.9 h1:4Z53AzfCcI8XK46eeoSDabroVk1GS2An04WDjNfwmio= github.com/openziti/foundation/v2 v2.0.9/go.mod h1:Z0gnQsGJb/H/B+pixLtmCIpw1dBllD4QoT6LeD35WzA= github.com/openziti/identity v1.0.28 h1:0Rjss3aBT4a2iC8IEXxzj2kfld5LTSD3eEUQkx8Uy+E= @@ -755,8 +755,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= -github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c h1:NRoLoZvkBTKvR5gQLgA3e0hqjkY9u1wm+iOL45VN/qI= -github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b h1:0LFwY6Q3gMACTjAbMZBjXAqTOzOwFaj2Ld6cjeQ7Rig= +github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= github.com/pquerna/otp v1.2.0/go.mod h1:dkJfzwRKNiegxyNb54X/3fLwhCynbMspSyWKnvi1AEg= github.com/prometheus/client_golang v0.8.0/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= @@ -1019,8 +1019,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4= golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM= golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU= -golang.org/x/exp v0.0.0-20221211140036-ad323defaf05 h1:T8EldfGCcveFMewH5xAYxxoX3PSQMrsechlUGVFlQBU= -golang.org/x/exp v0.0.0-20221211140036-ad323defaf05/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc= +golang.org/x/exp v0.0.0-20221217163422-3c43f8badb15 h1:5oN1Pz/eDhCpbMbLstvIPa0b/BEQo6g6nwV3pLjfM6w= +golang.org/x/exp v0.0.0-20221217163422-3c43f8badb15/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc= golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js= golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0= golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0= diff --git a/ziti/cmd/api/ws.go b/ziti/cmd/api/ws.go index 5abcccb1d..9af7730e4 100644 --- a/ziti/cmd/api/ws.go +++ b/ziti/cmd/api/ws.go @@ -17,7 +17,6 @@ package api import ( - "fmt" "github.com/gorilla/websocket" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel/v2" @@ -43,7 +42,6 @@ func NewWsMgmtChannel(bindHandler channel.BindHandler) (channel.Channel, error) } wsUrl := strings.ReplaceAll(baseUrl, "http", "ws") + "/ws-api" - fmt.Println(wsUrl) tlsConfig, err := restClientIdentity.NewTlsClientConfig() if err != nil { return nil, err diff --git a/ziti/cmd/fabric/root.go b/ziti/cmd/fabric/root.go index ec26a7335..fdff1eff9 100644 --- a/ziti/cmd/fabric/root.go +++ b/ziti/cmd/fabric/root.go @@ -80,8 +80,7 @@ func newStreamCommand(p common.OptionsProvider) *cobra.Command { }, } - streamCmd.AddCommand(NewStreamMetricsCmd(p)) - streamCmd.AddCommand(NewStreamCircuitsCmd(p)) + streamCmd.AddCommand(NewStreamEventsCmd(p)) streamTracesCmd := NewStreamTracesCmd(p) streamCmd.AddCommand(streamTracesCmd) diff --git a/ziti/cmd/fabric/stream_circuits.go b/ziti/cmd/fabric/stream_circuits.go deleted file mode 100644 index bde0ae5af..000000000 --- a/ziti/cmd/fabric/stream_circuits.go +++ /dev/null @@ -1,94 +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 fabric - -import ( - "fmt" - "github.com/openziti/channel/v2" - "github.com/openziti/fabric/pb/mgmt_pb" - "github.com/openziti/ziti/ziti/cmd/api" - "github.com/openziti/ziti/ziti/cmd/common" - "github.com/spf13/cobra" - "google.golang.org/protobuf/proto" - "time" -) - -type streamCircuitsAction struct { - api.Options -} - -func NewStreamCircuitsCmd(p common.OptionsProvider) *cobra.Command { - action := streamCircuitsAction{ - Options: api.Options{ - CommonOptions: p(), - }, - } - - streamCircuitsCmd := &cobra.Command{ - Use: "circuits", - Short: "Stream circuit data as circuits are created/destroyed", - Args: cobra.MaximumNArgs(0), - Run: action.streamCircuits, - } - - action.AddCommonFlags(streamCircuitsCmd) - - return streamCircuitsCmd -} - -func (self *streamCircuitsAction) streamCircuits(*cobra.Command, []string) { - closeNotify := make(chan struct{}) - - bindHandler := func(binding channel.Binding) error { - binding.AddReceiveHandler(int32(mgmt_pb.ContentType_StreamCircuitsEventType), self) - binding.AddCloseHandler(channel.CloseHandlerF(func(ch channel.Channel) { - close(closeNotify) - })) - return nil - } - - ch, err := api.NewWsMgmtChannel(channel.BindHandlerF(bindHandler)) - if err != nil { - panic(err) - } - - requestMsg := channel.NewMessage(int32(mgmt_pb.ContentType_StreamCircuitsRequestType), nil) - if err = requestMsg.WithTimeout(5 * time.Second).SendAndWaitForWire(ch); err != nil { - panic(err) - } - - <-closeNotify -} - -func (self *streamCircuitsAction) HandleReceive(msg *channel.Message, _ channel.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/cmd/fabric/stream_events.go b/ziti/cmd/fabric/stream_events.go new file mode 100644 index 000000000..d0eb76ec1 --- /dev/null +++ b/ziti/cmd/fabric/stream_events.go @@ -0,0 +1,232 @@ +/* + 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 fabric + +import ( + "encoding/json" + "fmt" + "github.com/openziti/channel/v2" + "github.com/openziti/fabric/event" + "github.com/openziti/fabric/pb/mgmt_pb" + "github.com/openziti/ziti/ziti/cmd/api" + "github.com/openziti/ziti/ziti/cmd/common" + "github.com/pkg/errors" + "github.com/spf13/cobra" + "os" + "time" +) + +type streamEventsAction struct { + api.Options + all bool + apiSessions bool + circuits bool + entityCounts bool + links bool + metrics bool + routers bool + services bool + sessions bool + usage bool + + metricsSourceFilter string + metricsFilter string + entityCountsInterval time.Duration + usageVersion uint8 +} + +func NewStreamEventsCmd(p common.OptionsProvider) *cobra.Command { + action := streamEventsAction{ + Options: api.Options{ + CommonOptions: p(), + }, + } + + streamEventsCmd := &cobra.Command{ + Use: "events", + Short: "Stream events", + Example: "ziti fabric stream events --circuits --metrics --metrics-filter '.*'", + Args: cobra.ExactArgs(0), + RunE: action.streamEvents, + } + + action.AddCommonFlags(streamEventsCmd) + streamEventsCmd.Flags().BoolVar(&action.all, "all", false, "Include all events") + streamEventsCmd.Flags().BoolVar(&action.apiSessions, "api-sessions", false, "Include api-session events") + streamEventsCmd.Flags().BoolVar(&action.circuits, "circuits", false, "Include circuit events") + streamEventsCmd.Flags().BoolVar(&action.entityCounts, "entity-counts", false, "Include entity count events") + streamEventsCmd.Flags().BoolVar(&action.links, "links", false, "Include link events") + streamEventsCmd.Flags().BoolVar(&action.metrics, "metrics", false, "Include metrics events") + streamEventsCmd.Flags().BoolVar(&action.routers, "routers", false, "Include router events") + streamEventsCmd.Flags().BoolVar(&action.services, "services", false, "Include service events") + streamEventsCmd.Flags().BoolVar(&action.sessions, "sessions", false, "Include session events") + streamEventsCmd.Flags().BoolVar(&action.usage, "usage", false, "Include usage events") + streamEventsCmd.Flags().DurationVar(&action.entityCountsInterval, "entity-counts-interval", 5*time.Minute, "Specify the entity count event interval") + streamEventsCmd.Flags().StringVar(&action.metricsSourceFilter, "metrics-source-filter", "", "Specify which sources to stream metrics from") + streamEventsCmd.Flags().StringVar(&action.metricsFilter, "metrics-filter", "", "Specify which metrics to stream") + streamEventsCmd.Flags().Uint8Var(&action.usageVersion, "usage-version", 3, "Specify which version of usage data to stream. Valid versions: [2,3]") + return streamEventsCmd +} + +func (self *streamEventsAction) buildSubscriptions(cmd *cobra.Command) []*event.Subscription { + var subscriptions []*event.Subscription + + if self.apiSessions || (self.all && !cmd.Flags().Changed("api-sessions")) { + subscriptions = append(subscriptions, &event.Subscription{ + Type: "edge.apiSessions", + }) + } + + if self.circuits || (self.all && !cmd.Flags().Changed("circuits")) { + subscriptions = append(subscriptions, &event.Subscription{ + Type: "fabric.circuits", + }) + } + + if self.entityCounts || (self.all && !cmd.Flags().Changed("entity-counts")) { + subscription := &event.Subscription{ + Type: "edge.entityCounts", + } + if cmd.Flags().Changed("entity-counts-interval") { + subscription.Options = map[string]interface{}{ + "interval": self.entityCountsInterval.String(), + } + } + subscriptions = append(subscriptions, subscription) + } + + if self.links || (self.all && !cmd.Flags().Changed("links")) { + subscriptions = append(subscriptions, &event.Subscription{ + Type: "fabric.links", + }) + } + + if self.metrics || (self.all && !cmd.Flags().Changed("metrics")) { + subscription := &event.Subscription{ + Type: "metrics", + Options: map[string]interface{}{}, + } + + if cmd.Flags().Changed("metrics-source-filter") { + subscription.Options["sourceFilter"] = self.metricsSourceFilter + } + + if cmd.Flags().Changed("metrics-filter") { + subscription.Options["metricFilter"] = self.metricsFilter + } + + subscriptions = append(subscriptions, subscription) + } + + if self.routers || (self.all && !cmd.Flags().Changed("routers")) { + subscriptions = append(subscriptions, &event.Subscription{ + Type: "fabric.routers", + }) + } + + if self.sessions || (self.all && !cmd.Flags().Changed("services")) { + subscriptions = append(subscriptions, &event.Subscription{ + Type: "services", + }) + } + + if self.sessions || (self.all && !cmd.Flags().Changed("sessions")) { + subscriptions = append(subscriptions, &event.Subscription{ + Type: "edge.sessions", + }) + } + + if self.usage || (self.all && !cmd.Flags().Changed("usage")) { + subscription := &event.Subscription{ + Type: "fabric.usage", + Options: map[string]interface{}{ + "version": self.usageVersion, + }, + } + + subscriptions = append(subscriptions, subscription) + } + + return subscriptions +} + +func (self *streamEventsAction) streamEvents(cmd *cobra.Command, _ []string) error { + if self.usageVersion < 2 || self.usageVersion > 3 { + return errors.New("invalid usage version") + } + streamEventsRequest := map[string]interface{}{} + streamEventsRequest["format"] = "json" + + subscriptions := self.buildSubscriptions(cmd) + if len(subscriptions) == 0 { + self.all = true + subscriptions = self.buildSubscriptions(cmd) + } + + streamEventsRequest["subscriptions"] = subscriptions + + closeNotify := make(chan struct{}) + + bindHandler := func(binding channel.Binding) error { + binding.AddReceiveHandler(int32(mgmt_pb.ContentType_StreamEventsEventType), self) + binding.AddCloseHandler(channel.CloseHandlerF(func(ch channel.Channel) { + close(closeNotify) + })) + return nil + } + + ch, err := api.NewWsMgmtChannel(channel.BindHandlerF(bindHandler)) + if err != nil { + return err + } + + msgBytes, err := json.Marshal(streamEventsRequest) + if err != nil { + return err + } + + if self.Verbose { + fmt.Printf("Request: %v\n", string(msgBytes)) + } + + requestMsg := channel.NewMessage(int32(mgmt_pb.ContentType_StreamEventsRequestType), msgBytes) + responseMsg, err := requestMsg.WithTimeout(time.Duration(self.Timeout) * time.Second).SendForReply(ch) + if err != nil { + return err + } + + if responseMsg.ContentType == channel.ContentTypeResultType { + result := channel.UnmarshalResult(responseMsg) + if result.Success { + if self.Verbose { + fmt.Printf("event streaming started: %v\n", result.Message) + } + } else { + fmt.Printf("error starting event streaming [%s]\n", result.Message) + os.Exit(1) + } + } else { + return errors.Errorf("unexpected response type %v", responseMsg.ContentType) + } + + <-closeNotify + return nil +} + +func (self *streamEventsAction) HandleReceive(msg *channel.Message, _ channel.Channel) { + fmt.Println(string(msg.Body)) +} diff --git a/ziti/cmd/fabric/stream_metrics.go b/ziti/cmd/fabric/stream_metrics.go deleted file mode 100644 index bbb8082bf..000000000 --- a/ziti/cmd/fabric/stream_metrics.go +++ /dev/null @@ -1,135 +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 fabric - -import ( - "fmt" - "github.com/openziti/channel/v2" - "github.com/openziti/fabric/pb/mgmt_pb" - "github.com/openziti/ziti/ziti/cmd/api" - "github.com/openziti/ziti/ziti/cmd/common" - "github.com/spf13/cobra" - "google.golang.org/protobuf/proto" - "google.golang.org/protobuf/types/known/timestamppb" - "sort" - "time" -) - -type streamMetricsAction struct { - api.Options -} - -func NewStreamMetricsCmd(p common.OptionsProvider) *cobra.Command { - action := streamMetricsAction{ - Options: api.Options{ - CommonOptions: p(), - }, - } - - streamMetricsCmd := &cobra.Command{ - Use: "metrics ", - Short: "Stream fabric metrics", - Args: cobra.MaximumNArgs(2), - Run: action.streamMetrics, - } - - action.AddCommonFlags(streamMetricsCmd) - - return streamMetricsCmd -} - -func (self *streamMetricsAction) streamMetrics(_ *cobra.Command, args []string) { - closeNotify := make(chan struct{}) - - bindHandler := func(binding channel.Binding) error { - binding.AddReceiveHandler(int32(mgmt_pb.ContentType_StreamMetricsEventType), self) - binding.AddCloseHandler(channel.CloseHandlerF(func(ch channel.Channel) { - close(closeNotify) - })) - return nil - } - - ch, err := api.NewWsMgmtChannel(channel.BindHandlerF(bindHandler)) - if err != nil { - panic(err) - } - - var matchers []*mgmt_pb.StreamMetricsRequest_MetricMatcher - - if len(args) > 0 { - matcher := &mgmt_pb.StreamMetricsRequest_MetricMatcher{NameRegex: args[0]} - if len(args) > 1 { - matcher.SourceIDRegex = args[1] - } - matchers = append(matchers, matcher) - } - - request := &mgmt_pb.StreamMetricsRequest{Matchers: matchers} - body, err := proto.Marshal(request) - if err != nil { - panic(err) - } - - requestMsg := channel.NewMessage(int32(mgmt_pb.ContentType_StreamMetricsRequestType), body) - if err = requestMsg.WithTimeout(time.Duration(self.Timeout) * time.Second).SendAndWaitForWire(ch); err != nil { - panic(err) - } - - <-closeNotify -} - -func (self *streamMetricsAction) HandleReceive(msg *channel.Message, _ channel.Channel) { - response := &mgmt_pb.StreamMetricsEvent{} - err := proto.Unmarshal(msg.Body, response) - if err != nil { - panic(err) - } - - fmt.Printf("%v - source(%v)\n", self.format(response.Timestamp), response.SourceId) - fmt.Printf("\tTags: %v\n", response.Tags) - - var keys []string - var outputMap = make(map[string]string) - for name, value := range response.IntMetrics { - outputMap[name] = fmt.Sprintf("%v=%v", name, value) - keys = append(keys, name) - } - - for name, value := range response.FloatMetrics { - outputMap[name] = fmt.Sprintf("%v=%v", name, value) - keys = append(keys, name) - } - - sort.Strings(keys) - - for _, key := range keys { - fmt.Println(outputMap[key]) - } - - for _, bucket := range response.IntervalMetrics { - fmt.Printf("%v: (%v) -> (%v)\n", bucket.Name, self.format(bucket.IntervalStartUTC), self.format(bucket.IntervalEndUTC)) - for name, value := range bucket.Values { - fmt.Printf("\t%v=%v\n", name, value) - } - } - - fmt.Println() -} - -func (self *streamMetricsAction) format(protobufTS *timestamppb.Timestamp) string { - return protobufTS.AsTime().Format(time.RFC3339) -} diff --git a/ziti/router/run.go b/ziti/router/run.go index 9bab59655..958cffde6 100644 --- a/ziti/router/run.go +++ b/ziti/router/run.go @@ -87,7 +87,7 @@ func run(cmd *cobra.Command, args []string) { xgressEdgeTransportFactory := xgress_edge_transport.NewFactory() xgress.GlobalRegistry().Register(xgress_edge_transport.BindingName, xgressEdgeTransportFactory) - xgressEdgeTunnelFactory := xgress_edge_tunnel.NewFactory(config, stateManager) + xgressEdgeTunnelFactory := xgress_edge_tunnel.NewFactory(config, stateManager, r.GetMetricsRegistry()) xgress.GlobalRegistry().Register(edge_common.TunnelBinding, xgressEdgeTunnelFactory) if err := r.RegisterXrctrl(xgressEdgeTunnelFactory); err != nil { logrus.WithError(err).Panic("error registering edge tunnel in framework")