Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix nats reconnects #8880

Merged
merged 2 commits into from
Apr 18, 2024
Merged

Fix nats reconnects #8880

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
eb5035c
fix(natsjsregistry): fix reconnects
kobergj Apr 17, 2024
7db6300
fix(changelog): improve wording
kobergj Apr 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions changelog/unreleased/nats-reconnects.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
Bugfix: Nats reconnects

Natsjs kv registry could not handle reconnects correctly. This fixes it.
kobergj marked this conversation as resolved.
Show resolved Hide resolved

https://github.com/owncloud/ocis/pull/8880
38 changes: 34 additions & 4 deletions ocis-pkg/natsjsregistry/registry.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,10 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"strings"
"sync"
"time"

natsjskv "github.com/go-micro/plugins/v4/store/nats-js-kv"
Expand Down Expand Up @@ -36,27 +38,33 @@ func NewRegistry(opts ...registry.Option) registry.Registry {
o(&options)
}
exp, _ := options.Context.Value(expiryKey{}).(time.Duration)
return &storeregistry{
n := &storeregistry{
opts: options,
store: natsjskv.NewStore(storeOptions(options)...),
typ: _registryName,
expiry: exp,
}
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
}

// Init inits the registry
func (n *storeregistry) Init(opts ...registry.Option) error {
n.lock.Lock()
defer n.lock.Unlock()

for _, o := range opts {
o(&n.opts)
}
return n.store.Init(storeOptions(n.opts)...)
n.store = natsjskv.NewStore(n.storeOptions(n.opts)...)
return n.store.Init(n.storeOptions(n.opts)...)
}

// Options returns the configured options
Expand All @@ -66,6 +74,9 @@ func (n *storeregistry) Options() registry.Options {

// Register adds a service to the registry
func (n *storeregistry) Register(s *registry.Service, _ ...registry.RegisterOption) error {
n.lock.RLock()
defer n.lock.RUnlock()

if s == nil {
return errors.New("wont store nil service")
}
Expand All @@ -82,11 +93,17 @@ func (n *storeregistry) Register(s *registry.Service, _ ...registry.RegisterOpti

// Deregister removes a service from the registry
func (n *storeregistry) Deregister(s *registry.Service, _ ...registry.DeregisterOption) error {
n.lock.RLock()
defer n.lock.RUnlock()

return n.store.Delete(s.Name)
}

// GetService gets a specific service from the registry
func (n *storeregistry) GetService(s string, _ ...registry.GetOption) ([]*registry.Service, error) {
n.lock.RLock()
defer n.lock.RUnlock()

recs, err := n.store.Read(s)
if err != nil {
return nil, err
Expand All @@ -104,6 +121,9 @@ func (n *storeregistry) GetService(s string, _ ...registry.GetOption) ([]*regist

// ListServices lists all registered services
func (n *storeregistry) ListServices(...registry.ListOption) ([]*registry.Service, error) {
n.lock.RLock()
defer n.lock.RUnlock()

keys, err := n.store.List()
if err != nil {
return nil, err
Expand Down Expand Up @@ -132,7 +152,7 @@ func (n *storeregistry) String() string {
return n.typ
}

func storeOptions(opts registry.Options) []store.Option {
func (n *storeregistry) storeOptions(opts registry.Options) []store.Option {
storeoptions := []store.Option{
store.Database("service-registry"),
store.Table("service-registry"),
Expand All @@ -150,6 +170,16 @@ func storeOptions(opts registry.Options) []store.Option {
natsOptions := nats.GetDefaultOptions()
natsOptions.Name = "nats-js-kv-registry"
natsOptions.User, natsOptions.Password = getAuth()
natsOptions.ReconnectedCB = func(_ *nats.Conn) {
if err := n.Init(); err != nil {
fmt.Println("cannot reconnect to nats")
os.Exit(1)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💥 OUCH

I guess I understand why you're doing that. And I don't have a better solution at hand. But I still feel the need to say: This is ugly 😄

Would be nice if we could find a way for a more graceful error handling here.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes I 100% agree. ❤️ I'll fix it as soon as we have a proper solution.

}
}
natsOptions.ClosedCB = func(_ *nats.Conn) {
fmt.Println("nats connection closed")
os.Exit(1)
}
storeoptions = append(storeoptions, natsjskv.NatsOptions(natsOptions))

if so, ok := opts.Context.Value(storeOptionsKey{}).([]store.Option); ok {
Expand Down