// Package cache provides a registry cache package cache import ( "math" "math/rand" "sync" "time" "github.com/unistack-org/micro/v3/logger" "github.com/unistack-org/micro/v3/registry" util "github.com/unistack-org/micro/v3/util/registry" ) // Cache is the registry cache interface type Cache interface { // embed the registry interface registry.Registry // stop the cache watcher Stop() } type Options struct { // TTL is the cache TTL TTL time.Duration } type Option func(o *Options) type cache struct { registry.Registry opts Options // registry cache. services,ttls,watched,running are grouped by doman sync.RWMutex services map[string]services ttls map[string]ttls watched map[string]watched running map[string]bool // used to stop the caches exit chan bool // indicate whether its running status of the registry used to hold onto the cache in failure state status error } type services map[string][]*registry.Service type ttls map[string]time.Time type watched map[string]bool var defaultTTL = time.Minute func backoff(attempts int) time.Duration { if attempts == 0 { return time.Duration(0) } return time.Duration(math.Pow(10, float64(attempts))) * time.Millisecond } func (c *cache) getStatus() error { c.RLock() defer c.RUnlock() return c.status } func (c *cache) setStatus(err error) { c.Lock() c.status = err c.Unlock() } // isValid checks if the service is valid func (c *cache) isValid(services []*registry.Service, ttl time.Time) bool { // no services exist if len(services) == 0 { return false } // ttl is invalid if ttl.IsZero() { return false } // time since ttl is longer than timeout if time.Since(ttl) > 0 { return false } // ok return true } func (c *cache) quit() bool { select { case <-c.exit: return true default: return false } } func (c *cache) del(domain, service string) { // don't blow away cache in error state if err := c.getStatus(); err != nil { return } c.Lock() defer c.Unlock() if _, ok := c.services[domain]; ok { delete(c.services[domain], service) } if _, ok := c.ttls[domain]; ok { delete(c.ttls[domain], service) } } func (c *cache) get(domain, service string) ([]*registry.Service, error) { var services []*registry.Service var ttl time.Time // lookup the values in the cache before calling the underlying registrry c.RLock() if srvs, ok := c.services[domain]; ok { services = srvs[service] } if tt, ok := c.ttls[domain]; ok { ttl = tt[service] } c.RUnlock() // got services && within ttl so return a copy of the services if c.isValid(services, ttl) { return util.Copy(services), nil } // get does the actual request for a service and cache it get := func(domain string, service string, cached []*registry.Service) ([]*registry.Service, error) { // ask the registry services, err := c.Registry.GetService(service, registry.GetDomain(domain)) if err != nil { // set the error status c.setStatus(err) // check the cache if len(cached) > 0 { return cached, nil } // otherwise return error return nil, err } // reset the status if err := c.getStatus(); err != nil { c.setStatus(nil) } // cache results c.set(domain, service, util.Copy(services)) return services, nil } // watch service if not watched c.RLock() var ok bool if _, d := c.watched[domain]; d { if _, s := c.watched[domain][service]; s { ok = true } } c.RUnlock() // check if its being watched if !ok { c.Lock() // add domain if not registered if _, ok := c.watched[domain]; !ok { c.watched[domain] = make(map[string]bool) } // set to watched c.watched[domain][service] = true running := c.running[domain] c.Unlock() // only kick it off if not running if !running { go c.run(domain) } } // get and return services return get(domain, service, services) } func (c *cache) set(domain string, service string, srvs []*registry.Service) { c.Lock() defer c.Unlock() if _, ok := c.services[domain]; !ok { c.services[domain] = make(services) } if _, ok := c.ttls[domain]; !ok { c.ttls[domain] = make(ttls) } c.services[domain][service] = srvs c.ttls[domain][service] = time.Now().Add(c.opts.TTL) } func (c *cache) update(domain string, res *registry.Result) { if res == nil || res.Service == nil { return } // only save watched services since the service using the cache may only depend on a handful // of other services c.RLock() if _, ok := c.watched[res.Service.Name]; !ok { c.RUnlock() return } // we're not going to cache anything unless there was already a lookup services, ok := c.services[domain][res.Service.Name] if !ok { c.RUnlock() return } c.RUnlock() if len(res.Service.Nodes) == 0 { switch res.Action { case "delete": c.del(domain, res.Service.Name) } return } // existing service found var service *registry.Service var index int for i, s := range services { if s.Version == res.Service.Version { service = s index = i } } switch res.Action { case "create", "update": if service == nil { c.set(domain, res.Service.Name, append(services, res.Service)) return } // append old nodes to new service for _, cur := range service.Nodes { var seen bool for _, node := range res.Service.Nodes { if cur.Id == node.Id { seen = true break } } if !seen { res.Service.Nodes = append(res.Service.Nodes, cur) } } services[index] = res.Service c.set(domain, res.Service.Name, services) case "delete": if service == nil { return } var nodes []*registry.Node // filter cur nodes to remove the dead one for _, cur := range service.Nodes { var seen bool for _, del := range res.Service.Nodes { if del.Id == cur.Id { seen = true break } } if !seen { nodes = append(nodes, cur) } } // still got nodes, save and return if len(nodes) > 0 { service.Nodes = nodes services[index] = service c.set(domain, service.Name, services) return } // zero nodes left // only have one thing to delete // nuke the thing if len(services) == 1 { c.del(domain, service.Name) return } // still have more than 1 service // check the version and keep what we know var srvs []*registry.Service for _, s := range services { if s.Version != service.Version { srvs = append(srvs, s) } } // save c.set(domain, service.Name, srvs) } } // run starts the cache watcher loop // it creates a new watcher if there's a problem func (c *cache) run(domain string) { c.Lock() c.running[domain] = true c.Unlock() // reset watcher on exit defer func() { c.Lock() c.watched[domain] = make(map[string]bool) c.running[domain] = false c.Unlock() }() var a, b int for { // exit early if already dead if c.quit() { return } // jitter before starting j := rand.Int63n(100) time.Sleep(time.Duration(j) * time.Millisecond) // create new watcher w, err := c.Registry.Watch(registry.WatchDomain(domain)) if err != nil { if c.quit() { return } d := backoff(a) c.setStatus(err) if a > 3 { if logger.V(logger.TraceLevel) { logger.Trace("rcache: ", err, " backing off ", d) } a = 0 } time.Sleep(d) a++ continue } // reset a a = 0 // watch for events if err := c.watch(domain, w); err != nil { if c.quit() { return } d := backoff(b) c.setStatus(err) if b > 3 { if logger.V(logger.TraceLevel) { logger.Trace("rcache: ", err, " backing off ", d) } b = 0 } time.Sleep(d) b++ continue } // reset b b = 0 } } // watch loops the next event and calls update // it returns if there's an error func (c *cache) watch(domain string, w registry.Watcher) error { // used to stop the watch stop := make(chan bool) // manage this loop go func() { defer w.Stop() select { // wait for exit case <-c.exit: return // we've been stopped case <-stop: return } }() for { res, err := w.Next() if err != nil { close(stop) return err } // reset the error status since we succeeded if err := c.getStatus(); err != nil { // reset status c.setStatus(nil) } // for wildcard queries, the domain will be * and not the services domain, so we'll check to // see if it was provided in the metadata. dom := domain if res.Service.Metadata != nil && len(res.Service.Metadata["domain"]) > 0 { dom = res.Service.Metadata["domain"] } c.update(dom, res) } } func (c *cache) GetService(service string, opts ...registry.GetOption) ([]*registry.Service, error) { // parse the options, fallback to the default domain var options registry.GetOptions for _, o := range opts { o(&options) } if len(options.Domain) == 0 { options.Domain = registry.DefaultDomain } // get the service services, err := c.get(options.Domain, service) if err != nil { return nil, err } // if there's nothing return err if len(services) == 0 { return nil, registry.ErrNotFound } // return services return services, nil } func (c *cache) Stop() { c.Lock() defer c.Unlock() select { case <-c.exit: return default: close(c.exit) } } func (c *cache) String() string { return "cache" } // New returns a new cache func New(r registry.Registry, opts ...Option) Cache { rand.Seed(time.Now().UnixNano()) options := Options{ TTL: defaultTTL, } for _, o := range opts { o(&options) } return &cache{ Registry: r, opts: options, running: make(map[string]bool), watched: make(map[string]watched), services: make(map[string]services), ttls: make(map[string]ttls), exit: make(chan bool), } }