From 01fa63e7164f9a412730484a7bb289540a993f15 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Thu, 17 Feb 2022 10:41:38 -0500 Subject: [PATCH] Use updated store events. Fixes openziti/fabric#319 --- controller/network/network.go | 18 -------- controller/network/router.go | 86 ++++++++++++++++++++++++++++------- controller/network/service.go | 13 ++++++ 3 files changed, 82 insertions(+), 35 deletions(-) diff --git a/controller/network/network.go b/controller/network/network.go index dd381fdf8..382b66938 100644 --- a/controller/network/network.go +++ b/controller/network/network.go @@ -898,24 +898,6 @@ func (network *Network) showOptions() { } } -func (network *Network) GetServiceCache() Cache { - return network.Services -} - -func (network *Network) NotifyRouterRenamed(id, name string) { - if cached, _ := network.Routers.cache.Get(id); cached != nil { - if cachedRouter, ok := cached.(*Router); ok { - cachedRouter.Name = name - } - } - - if cached, _ := network.Routers.connected.Get(id); cached != nil { - if cachedRouter, ok := cached.(*Router); ok { - cachedRouter.Name = name - } - } -} - func (network *Network) Inspect(name string) *string { if strings.ToLower(name) == "stackdump" { result := debugz.GenerateStack() diff --git a/controller/network/router.go b/controller/network/router.go index 13e518aee..4b56b08cb 100644 --- a/controller/network/router.go +++ b/controller/network/router.go @@ -24,7 +24,6 @@ import ( "github.com/openziti/foundation/common" "github.com/openziti/foundation/storage/boltz" "github.com/openziti/foundation/util/concurrenz" - "github.com/openziti/foundation/util/stringz" "github.com/orcaman/concurrent-map" "github.com/pkg/errors" "go.etcd.io/bbolt" @@ -150,6 +149,27 @@ func newRouterController(controllers *Controllers) *RouterController { store: controllers.stores.Router, } result.impl = result + + controllers.stores.Router.AddListener(boltz.EventUpdate, func(i ...interface{}) { + for _, val := range i { + if router, ok := val.(*db.Router); ok { + result.UpdateCachedRouter(router.Id) + } else { + pfxlog.Logger().Errorf("error in router listener. expected *db.Router, got %T", val) + } + } + }) + + controllers.stores.Router.AddListener(boltz.EventDelete, func(i ...interface{}) { + for _, val := range i { + if router, ok := val.(*db.Router); ok { + result.HandleRouterDelete(router.Id) + } else { + pfxlog.Logger().Errorf("error in router listener. expected *db.Router, got %T", val) + } + } + }) + return result } @@ -232,23 +252,11 @@ func (ctrl *RouterController) readInTx(tx *bbolt.Tx, id string) (*Router, error) return entity, nil } -func (ctrl *RouterController) UpdateCachedFingerprint(id, fingerprint string) { - if val, ok := ctrl.cache.Get(id); ok { - if router, ok := val.(*Router); ok { - router.Fingerprint = &fingerprint - } else { - pfxlog.Logger().Errorf("encountered %t in router cache, expected *Router", val) - } - } -} - func (ctrl *RouterController) Update(r *Router) error { if err := ctrl.updateGeneral(r, nil); err != nil { return err } - ctrl.UpdateCachedFingerprint(r.Id, stringz.OrEmpty(r.Fingerprint)) - return nil } @@ -257,13 +265,57 @@ func (ctrl *RouterController) Patch(r *Router, checker boltz.FieldChecker) error return err } - if checker.IsUpdated("fingerprint") { - ctrl.UpdateCachedFingerprint(r.Id, stringz.OrEmpty(r.Fingerprint)) - } - return nil } +func (ctrl *RouterController) HandleRouterDelete(id string) { + log := pfxlog.Logger().WithField("routerId", id) + log.Debug("processing router delete") + ctrl.cache.Remove(id) + + // if we close the control channel, the router will get removed from the connected cache. We don't do it + // here because it results in deadlock + if v, found := ctrl.connected.Get(id); found { + if router, ok := v.(*Router); ok { + if ctrl := router.Control; ctrl != nil { + _ = ctrl.Close() + log.Warn("connected router deleted, disconnecting router") + } else { + log.Warn("deleted router in connected cache doesn't have a connected control channel") + } + } else { + log.Errorf("cached router of wrong type, expected *Router, was %T", v) + } + } else { + log.Debug("deleted router not connected, no further action required") + } +} + +func (ctrl *RouterController) UpdateCachedRouter(id string) { + log := pfxlog.Logger().WithField("routerId", id) + if router, err := ctrl.Read(id); err != nil { + log.WithError(err).Error("failed to read router for cache update") + } else { + upsertF := func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + if !exist { + return nil + } + + if cached, ok := valueInMap.(*Router); ok { + cached.Name = router.Name + cached.Fingerprint = router.Fingerprint + } else { + log.Errorf("cached router of wrong type, expected *Router, was %T", valueInMap) + } + + return valueInMap + } + + ctrl.cache.Upsert(id, router, upsertF) + ctrl.connected.Upsert(id, router, upsertF) + } +} + type RouterLinks struct { sync.Mutex allLinks atomic.Value diff --git a/controller/network/service.go b/controller/network/service.go index 94c68cf7e..e2ff23916 100644 --- a/controller/network/service.go +++ b/controller/network/service.go @@ -69,6 +69,19 @@ func newServiceController(controllers *Controllers) *ServiceController { } result.impl = result + cacheInvalidationF := func(i ...interface{}) { + for _, val := range i { + if service, ok := val.(*db.Service); ok { + result.RemoveFromCache(service.Id) + } else { + pfxlog.Logger().Errorf("error in service listener. expected *db.Service, got %T", val) + } + } + } + + controllers.stores.Service.AddListener(boltz.EventUpdate, cacheInvalidationF) + controllers.stores.Service.AddListener(boltz.EventDelete, cacheInvalidationF) + return result }