From 0c0866711c255af677716a2d045f2573db6f6afc Mon Sep 17 00:00:00 2001 From: jkoberg Date: Fri, 19 Jul 2024 11:33:26 +0200 Subject: [PATCH 1/3] fix(registry): Fix TTL and registration interval Signed-off-by: jkoberg --- changelog/unreleased/fix-natsjskv-registry.md | 1 + ocis-pkg/natsjsregistry/registry.go | 4 +++ ocis-pkg/registry/expiry.go | 33 +++++++++++++++++++ ocis-pkg/registry/registry.go | 14 +++++--- ocis-pkg/service/grpc/service.go | 5 ++- ocis-pkg/service/http/service.go | 5 ++- 6 files changed, 51 insertions(+), 11 deletions(-) create mode 100644 ocis-pkg/registry/expiry.go diff --git a/changelog/unreleased/fix-natsjskv-registry.md b/changelog/unreleased/fix-natsjskv-registry.md index 8e93a2856..04d1210ed 100644 --- a/changelog/unreleased/fix-natsjskv-registry.md +++ b/changelog/unreleased/fix-natsjskv-registry.md @@ -2,4 +2,5 @@ Bugfix: Repair nats-js-kv registry The registry would always send traffic to only one pod. This is now fixed and load should be spread evenly. Also implements watcher method so the cache can use it. +https://github.com/owncloud/ocis/pull/9654 https://github.com/owncloud/ocis/pull/9620 diff --git a/ocis-pkg/natsjsregistry/registry.go b/ocis-pkg/natsjsregistry/registry.go index 5a45fe52b..97094d757 100644 --- a/ocis-pkg/natsjsregistry/registry.go +++ b/ocis-pkg/natsjsregistry/registry.go @@ -180,6 +180,10 @@ func (n *storeregistry) storeOptions(opts registry.Options) []store.Option { natsjskv.EncodeKeys(), } + if ttl, ok := opts.Context.Value(expiryKey{}).(time.Duration); ok { + storeoptions = append(storeoptions, natsjskv.DefaultTTL(ttl)) + } + addr := []string{"127.0.0.1:9233"} if len(opts.Addrs) > 0 { addr = opts.Addrs diff --git a/ocis-pkg/registry/expiry.go b/ocis-pkg/registry/expiry.go new file mode 100644 index 000000000..0eb1c0212 --- /dev/null +++ b/ocis-pkg/registry/expiry.go @@ -0,0 +1,33 @@ +package registry + +import ( + "os" + "time" +) + +const ( + _registryRegisterIntervalEnv = "EXPERIMENTAL_REGISTER_INTERVAL" + _registryRegisterTTLEnv = "EXPERIMENTAL_REGISTER_TTL" + + // Note: _defaultRegisterInterval should always be lower than _defaultRegisterTTL + _defaultRegisterInterval = time.Second * 25 + _defaultRegisterTTL = time.Second * 30 +) + +// GetRegisterInterval returns the register interval from the environment. +func GetRegisterInterval() time.Duration { + d, err := time.ParseDuration(os.Getenv(_registryRegisterIntervalEnv)) + if err != nil { + return _defaultRegisterInterval + } + return d +} + +// GetRegisterTTL returns the register TTL from the environment. +func GetRegisterTTL() time.Duration { + d, err := time.ParseDuration(os.Getenv(_registryRegisterTTLEnv)) + if err != nil { + return _defaultRegisterTTL + } + return d +} diff --git a/ocis-pkg/registry/registry.go b/ocis-pkg/registry/registry.go index 1672a1a8e..130b3d630 100644 --- a/ocis-pkg/registry/registry.go +++ b/ocis-pkg/registry/registry.go @@ -31,11 +31,12 @@ var ( // Config is the config for a registry type Config struct { - Type string `mapstructure:"type"` - Addresses []string `mapstructure:"addresses"` - Username string `mapstructure:"username"` - Password string `mapstructure:"password"` - DisableCache bool `mapstructure:"disable_cache"` + Type string `mapstructure:"type"` + Addresses []string `mapstructure:"addresses"` + Username string `mapstructure:"username"` + Password string `mapstructure:"password"` + DisableCache bool `mapstructure:"disable_cache"` + TTL time.Duration `mapstructure:"ttl"` } // Option allows configuring the registry @@ -62,6 +63,7 @@ func GetRegistry(opts ...Option) mRegistry.Registry { case "natsjs", "nats-js", "nats-js-kv": // for backwards compatibility - we will stick with one of those _reg = natsjsregistry.NewRegistry( mRegistry.Addrs(cfg.Addresses...), + natsjsregistry.ServiceExpiry(cfg.TTL), ) case "memory": _reg = memr.NewRegistry() @@ -119,6 +121,8 @@ func getEnvs(opts ...Option) *Config { cfg.Addresses = s } + cfg.TTL = GetRegisterTTL() + for _, o := range opts { o(cfg) } diff --git a/ocis-pkg/service/grpc/service.go b/ocis-pkg/service/grpc/service.go index 5776c7995..3364c2194 100644 --- a/ocis-pkg/service/grpc/service.go +++ b/ocis-pkg/service/grpc/service.go @@ -4,7 +4,6 @@ import ( "crypto/tls" "fmt" "strings" - "time" mgrpcs "github.com/go-micro/plugins/v4/server/grpc" "github.com/go-micro/plugins/v4/wrapper/monitoring/prometheus" @@ -59,8 +58,8 @@ func NewServiceWithClient(client client.Client, opts ...Option) (Service, error) micro.Version(sopts.Version), micro.Context(sopts.Context), micro.Registry(registry.GetRegistry()), - micro.RegisterTTL(time.Second * 30), - micro.RegisterInterval(time.Second * 10), + micro.RegisterTTL(registry.GetRegisterTTL()), + micro.RegisterInterval(registry.GetRegisterInterval()), micro.WrapHandler(prometheus.NewHandlerWrapper()), micro.WrapClient(mtracer.NewClientWrapper( mtracer.WithTraceProvider(sopts.TraceProvider), diff --git a/ocis-pkg/service/http/service.go b/ocis-pkg/service/http/service.go index 8f4f66994..cc37ace2b 100644 --- a/ocis-pkg/service/http/service.go +++ b/ocis-pkg/service/http/service.go @@ -4,7 +4,6 @@ import ( "crypto/tls" "fmt" "strings" - "time" "github.com/owncloud/ocis/v2/ocis-pkg/broker" "github.com/owncloud/ocis/v2/ocis-pkg/registry" @@ -65,8 +64,8 @@ func NewService(opts ...Option) (Service, error) { micro.Context(sopts.Context), micro.Flags(sopts.Flags...), micro.Registry(registry.GetRegistry()), - micro.RegisterTTL(time.Second * 30), - micro.RegisterInterval(time.Second * 10), + micro.RegisterTTL(registry.GetRegisterTTL()), + micro.RegisterInterval(registry.GetRegisterInterval()), micro.WrapClient(mtracer.NewClientWrapper( mtracer.WithTraceProvider(sopts.TraceProvider), )), From 0c6ca19602708f9f36129f62febeee5a30a08595 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Fri, 19 Jul 2024 12:37:08 +0200 Subject: [PATCH 2/3] naming is hard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Jörn Friedrich Dreyer --- ocis-pkg/natsjsregistry/options.go | 8 ++++---- ocis-pkg/natsjsregistry/registry.go | 32 +++++++++++++++++------------ ocis-pkg/registry/registry.go | 6 +++--- 3 files changed, 26 insertions(+), 20 deletions(-) diff --git a/ocis-pkg/natsjsregistry/options.go b/ocis-pkg/natsjsregistry/options.go index dfcda616a..533d0bd87 100644 --- a/ocis-pkg/natsjsregistry/options.go +++ b/ocis-pkg/natsjsregistry/options.go @@ -9,7 +9,7 @@ import ( ) type storeOptionsKey struct{} -type expiryKey struct{} +type defaultTTLKey struct{} // StoreOptions sets the options for the underlying store func StoreOptions(opts []store.Option) registry.Option { @@ -21,12 +21,12 @@ func StoreOptions(opts []store.Option) registry.Option { } } -// ServiceExpiry allows setting an expiry time for service registrations -func ServiceExpiry(t time.Duration) registry.Option { +// DefaultTTL allows setting a default register TTL for services +func DefaultTTL(t time.Duration) registry.Option { return func(o *registry.Options) { if o.Context == nil { o.Context = context.Background() } - o.Context = context.WithValue(o.Context, expiryKey{}, t) + o.Context = context.WithValue(o.Context, defaultTTLKey{}, t) } } diff --git a/ocis-pkg/natsjsregistry/registry.go b/ocis-pkg/natsjsregistry/registry.go index 97094d757..d9dcd5f93 100644 --- a/ocis-pkg/natsjsregistry/registry.go +++ b/ocis-pkg/natsjsregistry/registry.go @@ -40,22 +40,22 @@ func NewRegistry(opts ...registry.Option) registry.Registry { for _, o := range opts { o(&options) } - exp, _ := options.Context.Value(expiryKey{}).(time.Duration) + defaultTTL, _ := options.Context.Value(defaultTTLKey{}).(time.Duration) n := &storeregistry{ - opts: options, - typ: _registryName, - expiry: exp, + opts: options, + typ: _registryName, + defaultTTL: defaultTTL, } n.store = natsjskv.NewStore(n.storeOptions(options)...) return n } type storeregistry struct { - opts registry.Options - store store.Store - typ string - expiry time.Duration - lock sync.RWMutex + opts registry.Options + store store.Store + typ string + defaultTTL time.Duration + lock sync.RWMutex } // Init inits the registry @@ -76,7 +76,7 @@ func (n *storeregistry) Options() registry.Options { } // Register adds a service to the registry -func (n *storeregistry) Register(s *registry.Service, _ ...registry.RegisterOption) error { +func (n *storeregistry) Register(s *registry.Service, opts ...registry.RegisterOption) error { n.lock.RLock() defer n.lock.RUnlock() @@ -84,6 +84,12 @@ func (n *storeregistry) Register(s *registry.Service, _ ...registry.RegisterOpti return errors.New("wont store nil service") } + var options registry.RegisterOptions + options.TTL = n.defaultTTL + for _, o := range opts { + o(&options) + } + unique := uuid.New().String() if s.Metadata == nil { s.Metadata = make(map[string]string) @@ -97,7 +103,7 @@ func (n *storeregistry) Register(s *registry.Service, _ ...registry.RegisterOpti return n.store.Write(&store.Record{ Key: s.Name + _serviceDelimiter + unique, Value: b, - Expiry: n.expiry, + Expiry: options.TTL, }) } @@ -180,8 +186,8 @@ func (n *storeregistry) storeOptions(opts registry.Options) []store.Option { natsjskv.EncodeKeys(), } - if ttl, ok := opts.Context.Value(expiryKey{}).(time.Duration); ok { - storeoptions = append(storeoptions, natsjskv.DefaultTTL(ttl)) + if defaultTTL, ok := opts.Context.Value(defaultTTLKey{}).(time.Duration); ok { + storeoptions = append(storeoptions, natsjskv.DefaultTTL(defaultTTL)) } addr := []string{"127.0.0.1:9233"} diff --git a/ocis-pkg/registry/registry.go b/ocis-pkg/registry/registry.go index 130b3d630..d8df986ad 100644 --- a/ocis-pkg/registry/registry.go +++ b/ocis-pkg/registry/registry.go @@ -36,7 +36,7 @@ type Config struct { Username string `mapstructure:"username"` Password string `mapstructure:"password"` DisableCache bool `mapstructure:"disable_cache"` - TTL time.Duration `mapstructure:"ttl"` + RegisterTTL time.Duration `mapstructure:"register_ttl"` } // Option allows configuring the registry @@ -63,7 +63,7 @@ func GetRegistry(opts ...Option) mRegistry.Registry { case "natsjs", "nats-js", "nats-js-kv": // for backwards compatibility - we will stick with one of those _reg = natsjsregistry.NewRegistry( mRegistry.Addrs(cfg.Addresses...), - natsjsregistry.ServiceExpiry(cfg.TTL), + natsjsregistry.DefaultTTL(cfg.RegisterTTL), ) case "memory": _reg = memr.NewRegistry() @@ -121,7 +121,7 @@ func getEnvs(opts ...Option) *Config { cfg.Addresses = s } - cfg.TTL = GetRegisterTTL() + cfg.RegisterTTL = GetRegisterTTL() for _, o := range opts { o(cfg) From c421507f1e80d2361e9b626e9c8a004ea9a52848 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Fri, 19 Jul 2024 12:55:16 +0200 Subject: [PATCH 3/3] reva services should respect same register ttl MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Jörn Friedrich Dreyer --- ocis-pkg/registry/register.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ocis-pkg/registry/register.go b/ocis-pkg/registry/register.go index 692307dd2..14ffbee94 100644 --- a/ocis-pkg/registry/register.go +++ b/ocis-pkg/registry/register.go @@ -16,12 +16,12 @@ func RegisterService(ctx context.Context, service *mRegistry.Service, logger log logger.Info().Msgf("registering external service %v@%v", node.Id, node.Address) - rOpts := []mRegistry.RegisterOption{mRegistry.RegisterTTL(time.Minute)} + rOpts := []mRegistry.RegisterOption{mRegistry.RegisterTTL(GetRegisterTTL())} if err := registry.Register(service, rOpts...); err != nil { logger.Fatal().Err(err).Msgf("Registration error for external service %v", service.Name) } - t := time.NewTicker(time.Second * 30) + t := time.NewTicker(GetRegisterInterval()) go func() { for {