extern/zrok
1
1
Fork 0
mirror of https://github.com/openziti/zrok.git synced 2025-02-24 14:11:37 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #339 from openziti/v0.4.0_limits_agent_api_usage

Browse Source 
Fix for Edge Management API Usage in Limits Agent
This commit is contained in:
Michael Quigley 2023-06-05 16:27:02 -04:00 committed by GitHub
commit a2df30a8bd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 82 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
controller/limits/accountLimitAction.go
View File

@ -2,7 +2,6 @@ package limits
import ( import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/openziti/edge-api/rest_management_api_client"
"github.com/openziti/zrok/controller/store" "github.com/openziti/zrok/controller/store"
"github.com/openziti/zrok/controller/zrokEdgeSdk" "github.com/openziti/zrok/controller/zrokEdgeSdk"
"github.com/pkg/errors" "github.com/pkg/errors"
@ -11,14 +10,14 @@ import (
type accountLimitAction struct { type accountLimitAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement zCfg *zrokEdgeSdk.Config
} }
func newAccountLimitAction(str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *accountLimitAction { func newAccountLimitAction(str *store.Store, zCfg *zrokEdgeSdk.Config) *accountLimitAction {
return &accountLimitAction{str, edge} return &accountLimitAction{str, zCfg}
} }
func (a *accountLimitAction) HandleAccount(acct *store.Account, rxBytes, txBytes int64, limit *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *accountLimitAction) HandleAccount(acct *store.Account, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error {
logrus.Infof("limiting '%v'", acct.Email) logrus.Infof("limiting '%v'", acct.Email)
envs, err := a.str.FindEnvironmentsForAccount(acct.Id, trx) envs, err := a.str.FindEnvironmentsForAccount(acct.Id, trx)
@ -26,6 +25,11 @@ func (a *accountLimitAction) HandleAccount(acct *store.Account, rxBytes, txBytes
return errors.Wrapf(err, "error finding environments for account '%v'", acct.Email) return errors.Wrapf(err, "error finding environments for account '%v'", acct.Email)
} }
edge, err := zrokEdgeSdk.Client(a.zCfg)
if err != nil {
return err
}
for _, env := range envs { for _, env := range envs {
shrs, err := a.str.FindSharesForEnvironment(env.Id, trx) shrs, err := a.str.FindSharesForEnvironment(env.Id, trx)
if err != nil { if err != nil {
@ -33,7 +37,7 @@ func (a *accountLimitAction) HandleAccount(acct *store.Account, rxBytes, txBytes
} }
for _, shr := range shrs { for _, shr := range shrs {
if err := zrokEdgeSdk.DeleteServicePoliciesDial(env.ZId, shr.Token, a.edge); err != nil { if err := zrokEdgeSdk.DeleteServicePoliciesDial(env.ZId, shr.Token, edge); err != nil {
return errors.Wrapf(err, "error deleting dial service policy for '%v'", shr.Token) return errors.Wrapf(err, "error deleting dial service policy for '%v'", shr.Token)
} }
logrus.Infof("removed dial service policy for share '%v' of environment '%v'", shr.Token, env.ZId) logrus.Infof("removed dial service policy for share '%v' of environment '%v'", shr.Token, env.ZId)

17
controller/limits/accountRelaxAction.go
View File

@ -2,19 +2,19 @@ package limits
import ( import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/openziti/edge-api/rest_management_api_client"
"github.com/openziti/zrok/controller/store" "github.com/openziti/zrok/controller/store"
"github.com/openziti/zrok/controller/zrokEdgeSdk"
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
) )
type accountRelaxAction struct { type accountRelaxAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement zCfg *zrokEdgeSdk.Config
} }
func newAccountRelaxAction(str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *accountRelaxAction { func newAccountRelaxAction(str *store.Store, zCfg *zrokEdgeSdk.Config) *accountRelaxAction {
return &accountRelaxAction{str, edge} return &accountRelaxAction{str, zCfg}
} }
func (a *accountRelaxAction) HandleAccount(acct *store.Account, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *accountRelaxAction) HandleAccount(acct *store.Account, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error {
@ -25,6 +25,11 @@ func (a *accountRelaxAction) HandleAccount(acct *store.Account, _, _ int64, _ *B
return errors.Wrapf(err, "error finding environments for account '%v'", acct.Email) return errors.Wrapf(err, "error finding environments for account '%v'", acct.Email)
} }
edge, err := zrokEdgeSdk.Client(a.zCfg)
if err != nil {
return err
}
for _, env := range envs { for _, env := range envs {
shrs, err := a.str.FindSharesForEnvironment(env.Id, trx) shrs, err := a.str.FindSharesForEnvironment(env.Id, trx)
if err != nil { if err != nil {
@ -34,11 +39,11 @@ func (a *accountRelaxAction) HandleAccount(acct *store.Account, _, _ int64, _ *B
for _, shr := range shrs { for _, shr := range shrs {
switch shr.ShareMode { switch shr.ShareMode {
case "public": case "public":
if err := relaxPublicShare(a.str, a.edge, shr, trx); err != nil { if err := relaxPublicShare(a.str, edge, shr, trx); err != nil {
return errors.Wrap(err, "error relaxing public share") return errors.Wrap(err, "error relaxing public share")
} }
case "private": case "private":
if err := relaxPrivateShare(a.str, a.edge, shr, trx); err != nil { if err := relaxPrivateShare(a.str, edge, shr, trx); err != nil {
return errors.Wrap(err, "error relaxing private share") return errors.Wrap(err, "error relaxing private share")
} }
} }

12
controller/limits/accountWarningAction.go
View File

@ -2,7 +2,6 @@ package limits
import ( import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/openziti/edge-api/rest_management_api_client"
"github.com/openziti/zrok/controller/emailUi" "github.com/openziti/zrok/controller/emailUi"
"github.com/openziti/zrok/controller/store" "github.com/openziti/zrok/controller/store"
"github.com/openziti/zrok/util" "github.com/openziti/zrok/util"
@ -11,16 +10,15 @@ import (
) )
type accountWarningAction struct { type accountWarningAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement cfg *emailUi.Config
cfg *emailUi.Config
} }
func newAccountWarningAction(cfg *emailUi.Config, str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *accountWarningAction { func newAccountWarningAction(cfg *emailUi.Config, str *store.Store) *accountWarningAction {
return &accountWarningAction{str, edge, cfg} return &accountWarningAction{str, cfg}
} }
func (a *accountWarningAction) HandleAccount(acct *store.Account, rxBytes, txBytes int64, limit *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *accountWarningAction) HandleAccount(acct *store.Account, rxBytes, txBytes int64, limit *BandwidthPerPeriod, _ *sqlx.Tx) error {
logrus.Infof("warning '%v'", acct.Email) logrus.Infof("warning '%v'", acct.Email)
if a.cfg != nil { if a.cfg != nil {

22
controller/limits/agent.go
View File

@ -34,25 +34,21 @@ type Agent struct {
} }
func NewAgent(cfg *Config, ifxCfg *metrics.InfluxConfig, zCfg *zrokEdgeSdk.Config, emailCfg *emailUi.Config, str *store.Store) (*Agent, error) { func NewAgent(cfg *Config, ifxCfg *metrics.InfluxConfig, zCfg *zrokEdgeSdk.Config, emailCfg *emailUi.Config, str *store.Store) (*Agent, error) {
edge, err := zrokEdgeSdk.Client(zCfg)
if err != nil {
return nil, err
}
a := &Agent{ a := &Agent{
cfg: cfg, cfg: cfg,
ifx: newInfluxReader(ifxCfg), ifx: newInfluxReader(ifxCfg),
zCfg: zCfg, zCfg: zCfg,
str: str, str: str,
queue: make(chan *metrics.Usage, 1024), queue: make(chan *metrics.Usage, 1024),
acctWarningActions: []AccountAction{newAccountWarningAction(emailCfg, str, edge)}, acctWarningActions: []AccountAction{newAccountWarningAction(emailCfg, str)},
acctLimitActions: []AccountAction{newAccountLimitAction(str, edge)}, acctLimitActions: []AccountAction{newAccountLimitAction(str, zCfg)},
acctRelaxActions: []AccountAction{newAccountRelaxAction(str, edge)}, acctRelaxActions: []AccountAction{newAccountRelaxAction(str, zCfg)},
envWarningActions: []EnvironmentAction{newEnvironmentWarningAction(emailCfg, str, edge)}, envWarningActions: []EnvironmentAction{newEnvironmentWarningAction(emailCfg, str)},
envLimitActions: []EnvironmentAction{newEnvironmentLimitAction(str, edge)}, envLimitActions: []EnvironmentAction{newEnvironmentLimitAction(str, zCfg)},
envRelaxActions: []EnvironmentAction{newEnvironmentRelaxAction(str, edge)}, envRelaxActions: []EnvironmentAction{newEnvironmentRelaxAction(str, zCfg)},
shrWarningActions: []ShareAction{newShareWarningAction(emailCfg, str, edge)}, shrWarningActions: []ShareAction{newShareWarningAction(emailCfg, str)},
shrLimitActions: []ShareAction{newShareLimitAction(str, edge)}, shrLimitActions: []ShareAction{newShareLimitAction(str, zCfg)},
shrRelaxActions: []ShareAction{newShareRelaxAction(str, edge)}, shrRelaxActions: []ShareAction{newShareRelaxAction(str, zCfg)},
close: make(chan struct{}), close: make(chan struct{}),
join: make(chan struct{}), join: make(chan struct{}),
} }

14
controller/limits/environmentLimitAction.go
View File

@ -2,7 +2,6 @@ package limits
import ( import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/openziti/edge-api/rest_management_api_client"
"github.com/openziti/zrok/controller/store" "github.com/openziti/zrok/controller/store"
"github.com/openziti/zrok/controller/zrokEdgeSdk" "github.com/openziti/zrok/controller/zrokEdgeSdk"
"github.com/pkg/errors" "github.com/pkg/errors"
@ -11,11 +10,11 @@ import (
type environmentLimitAction struct { type environmentLimitAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement zCfg *zrokEdgeSdk.Config
} }
func newEnvironmentLimitAction(str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *environmentLimitAction { func newEnvironmentLimitAction(str *store.Store, zCfg *zrokEdgeSdk.Config) *environmentLimitAction {
return &environmentLimitAction{str, edge} return &environmentLimitAction{str, zCfg}
} }
func (a *environmentLimitAction) HandleEnvironment(env *store.Environment, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *environmentLimitAction) HandleEnvironment(env *store.Environment, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error {
@ -26,8 +25,13 @@ func (a *environmentLimitAction) HandleEnvironment(env *store.Environment, _, _
return errors.Wrapf(err, "error finding shares for environment '%v'", env.ZId) return errors.Wrapf(err, "error finding shares for environment '%v'", env.ZId)
} }
edge, err := zrokEdgeSdk.Client(a.zCfg)
if err != nil {
return err
}
for _, shr := range shrs { for _, shr := range shrs {
if err := zrokEdgeSdk.DeleteServicePoliciesDial(env.ZId, shr.Token, a.edge); err != nil { if err := zrokEdgeSdk.DeleteServicePoliciesDial(env.ZId, shr.Token, edge); err != nil {
return errors.Wrapf(err, "error deleting dial service policy for '%v'", shr.Token) return errors.Wrapf(err, "error deleting dial service policy for '%v'", shr.Token)
} }
logrus.Infof("removed dial service policy for share '%v' of environment '%v'", shr.Token, env.ZId) logrus.Infof("removed dial service policy for share '%v' of environment '%v'", shr.Token, env.ZId)

17
controller/limits/environmentRelaxAction.go
View File

@ -2,19 +2,19 @@ package limits
import ( import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/openziti/edge-api/rest_management_api_client"
"github.com/openziti/zrok/controller/store" "github.com/openziti/zrok/controller/store"
"github.com/openziti/zrok/controller/zrokEdgeSdk"
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
) )
type environmentRelaxAction struct { type environmentRelaxAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement zCfg *zrokEdgeSdk.Config
} }
func newEnvironmentRelaxAction(str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *environmentRelaxAction { func newEnvironmentRelaxAction(str *store.Store, zCfg *zrokEdgeSdk.Config) *environmentRelaxAction {
return &environmentRelaxAction{str, edge} return &environmentRelaxAction{str, zCfg}
} }
func (a *environmentRelaxAction) HandleEnvironment(env *store.Environment, rxBytes, txBytes int64, limit *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *environmentRelaxAction) HandleEnvironment(env *store.Environment, rxBytes, txBytes int64, limit *BandwidthPerPeriod, trx *sqlx.Tx) error {
@ -25,15 +25,20 @@ func (a *environmentRelaxAction) HandleEnvironment(env *store.Environment, rxByt
return errors.Wrapf(err, "error finding shares for environment '%v'", env.ZId) return errors.Wrapf(err, "error finding shares for environment '%v'", env.ZId)
} }
edge, err := zrokEdgeSdk.Client(a.zCfg)
if err != nil {
return err
}
for _, shr := range shrs { for _, shr := range shrs {
if !shr.Deleted { if !shr.Deleted {
switch shr.ShareMode { switch shr.ShareMode {
case "public": case "public":
if err := relaxPublicShare(a.str, a.edge, shr, trx); err != nil { if err := relaxPublicShare(a.str, edge, shr, trx); err != nil {
return err return err
} }
case "private": case "private":
if err := relaxPrivateShare(a.str, a.edge, shr, trx); err != nil { if err := relaxPrivateShare(a.str, edge, shr, trx); err != nil {
return err return err
} }
} }

10
controller/limits/environmentWarningAction.go
View File

@ -2,7 +2,6 @@ package limits
import ( import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/openziti/edge-api/rest_management_api_client"
"github.com/openziti/zrok/controller/emailUi" "github.com/openziti/zrok/controller/emailUi"
"github.com/openziti/zrok/controller/store" "github.com/openziti/zrok/controller/store"
"github.com/openziti/zrok/util" "github.com/openziti/zrok/util"
@ -11,13 +10,12 @@ import (
) )
type environmentWarningAction struct { type environmentWarningAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement cfg *emailUi.Config
cfg *emailUi.Config
} }
func newEnvironmentWarningAction(cfg *emailUi.Config, str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *environmentWarningAction { func newEnvironmentWarningAction(cfg *emailUi.Config, str *store.Store) *environmentWarningAction {
return &environmentWarningAction{str, edge, cfg} return &environmentWarningAction{str, cfg}
} }
func (a *environmentWarningAction) HandleEnvironment(env *store.Environment, rxBytes, txBytes int64, limit *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *environmentWarningAction) HandleEnvironment(env *store.Environment, rxBytes, txBytes int64, limit *BandwidthPerPeriod, trx *sqlx.Tx) error {

14
controller/limits/shareLimitAction.go
View File

@ -2,7 +2,6 @@ package limits
import ( import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/openziti/edge-api/rest_management_api_client"
"github.com/openziti/zrok/controller/store" "github.com/openziti/zrok/controller/store"
"github.com/openziti/zrok/controller/zrokEdgeSdk" "github.com/openziti/zrok/controller/zrokEdgeSdk"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
@ -10,11 +9,11 @@ import (
type shareLimitAction struct { type shareLimitAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement zCfg *zrokEdgeSdk.Config
} }
func newShareLimitAction(str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *shareLimitAction { func newShareLimitAction(str *store.Store, zCfg *zrokEdgeSdk.Config) *shareLimitAction {
return &shareLimitAction{str, edge} return &shareLimitAction{str, zCfg}
} }
func (a *shareLimitAction) HandleShare(shr *store.Share, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *shareLimitAction) HandleShare(shr *store.Share, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error {
@ -25,7 +24,12 @@ func (a *shareLimitAction) HandleShare(shr *store.Share, _, _ int64, _ *Bandwidt
return err return err
} }
if err := zrokEdgeSdk.DeleteServicePoliciesDial(env.ZId, shr.Token, a.edge); err != nil { edge, err := zrokEdgeSdk.Client(a.zCfg)
if err != nil {
return err
}
if err := zrokEdgeSdk.DeleteServicePoliciesDial(env.ZId, shr.Token, edge); err != nil {
return err return err
} }
logrus.Infof("removed dial service policy for '%v'", shr.Token) logrus.Infof("removed dial service policy for '%v'", shr.Token)

15
controller/limits/shareRelaxAction.go
View File

@ -11,24 +11,29 @@ import (
type shareRelaxAction struct { type shareRelaxAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement zCfg *zrokEdgeSdk.Config
} }
func newShareRelaxAction(str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *shareRelaxAction { func newShareRelaxAction(str *store.Store, zCfg *zrokEdgeSdk.Config) *shareRelaxAction {
return &shareRelaxAction{str, edge} return &shareRelaxAction{str, zCfg}
} }
func (a *shareRelaxAction) HandleShare(shr *store.Share, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *shareRelaxAction) HandleShare(shr *store.Share, _, _ int64, _ *BandwidthPerPeriod, trx *sqlx.Tx) error {
logrus.Infof("relaxing '%v'", shr.Token) logrus.Infof("relaxing '%v'", shr.Token)
if !shr.Deleted { if !shr.Deleted {
edge, err := zrokEdgeSdk.Client(a.zCfg)
if err != nil {
return err
}
switch shr.ShareMode { switch shr.ShareMode {
case "public": case "public":
if err := relaxPublicShare(a.str, a.edge, shr, trx); err != nil { if err := relaxPublicShare(a.str, edge, shr, trx); err != nil {
return err return err
} }
case "private": case "private":
if err := relaxPrivateShare(a.str, a.edge, shr, trx); err != nil { if err := relaxPrivateShare(a.str, edge, shr, trx); err != nil {
return err return err
} }
} }

10
controller/limits/shareWarningAction.go
View File

@ -2,7 +2,6 @@ package limits
import ( import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/openziti/edge-api/rest_management_api_client"
"github.com/openziti/zrok/controller/emailUi" "github.com/openziti/zrok/controller/emailUi"
"github.com/openziti/zrok/controller/store" "github.com/openziti/zrok/controller/store"
"github.com/openziti/zrok/util" "github.com/openziti/zrok/util"
@ -11,13 +10,12 @@ import (
) )
type shareWarningAction struct { type shareWarningAction struct {
str *store.Store str *store.Store
edge *rest_management_api_client.ZitiEdgeManagement cfg *emailUi.Config
cfg *emailUi.Config
} }
func newShareWarningAction(cfg *emailUi.Config, str *store.Store, edge *rest_management_api_client.ZitiEdgeManagement) *shareWarningAction { func newShareWarningAction(cfg *emailUi.Config, str *store.Store) *shareWarningAction {
return &shareWarningAction{str, edge, cfg} return &shareWarningAction{str, cfg}
} }
func (a *shareWarningAction) HandleShare(shr *store.Share, rxBytes, txBytes int64, limit *BandwidthPerPeriod, trx *sqlx.Tx) error { func (a *shareWarningAction) HandleShare(shr *store.Share, rxBytes, txBytes int64, limit *BandwidthPerPeriod, trx *sqlx.Tx) error {