Allow to configure multiple rpc endpoints (#970)

CHANGELOG.md

@@ -4,6 +4,23 @@ This document outlines major changes between releases.
 
 ## [Unreleased]
 
+### Changed
+- Single `rpc_endpoint` config option replaced with multiple endpoints `fschain.endpoints` option (#968)
+
+### Updating from 0.30.1
+
+Notice that the configuration parameter `fschain.endpoints` contains multiple endpoints.
+For migration, you should update config from old notation
+```yaml
+rpc_endpoint: http://some-host:30333
+```
+to the new one
+```yaml
+fschain:
+  endpoints:
+    - http://some-host:30333
+```
+
 ## [0.30.1] - 2024-06-19
 
 ### Added

api/resolver/init.go

@@ -30,8 +30,8 @@ func (r *Container) ResolveCID(ctx context.Context, name string) (cid.ID, error)
 }
 
 // UpdateResolvers allows to update resolver in runtime. Resolvers will be created from scratch.
-func (r *Container) UpdateResolvers(ctx context.Context, endpoint string) error {
-	newResolver, err := NewResolver(ctx, endpoint)
+func (r *Container) UpdateResolvers(ctx context.Context, endpoints []string) error {
+	newResolver, err := NewResolver(ctx, endpoints)
 	if err != nil {
 		return fmt.Errorf("resolver reinit: %w", err)
 	}
@@ -44,8 +44,8 @@ func (r *Container) UpdateResolvers(ctx context.Context, endpoint string) error
 }
 
 // NewContainer is a constructor for the [Container].
-func NewContainer(ctx context.Context, endpoint string) (*Container, error) {
-	newResolver, err := NewResolver(ctx, endpoint)
+func NewContainer(ctx context.Context, endpoints []string) (*Container, error) {
+	newResolver, err := NewResolver(ctx, endpoints)
 	if err != nil {
 		return nil, fmt.Errorf("resolver reinit: %w", err)
 	}
@@ -55,26 +55,32 @@ func NewContainer(ctx context.Context, endpoint string) (*Container, error) {
 	}, nil
 }
 
-// NewResolver returns resolver depending on corresponding endpoint.
+// NewResolver returns resolver depending on corresponding endpoints.
 //
 // If endpoint is empty, error will be returned.
-func NewResolver(ctx context.Context, endpoint string) (*NNSResolver, error) {
-	if endpoint == "" {
-		return nil, errors.New("endpoint must be set")
+func NewResolver(ctx context.Context, endpoints []string) (*NNSResolver, error) {
+	if len(endpoints) == 0 {
+		return nil, errors.New("endpoints must be set")
 	}
 
-	cl, err := rpcClient(ctx, endpoint)
-	if err != nil {
-		return nil, fmt.Errorf("rpcclient: %w", err)
-	}
+	var readers = make([]*nns.ContractReader, 0, len(endpoints))
 
-	inv := invoker.New(cl, nil)
-	nnsReader, err := nns.NewInferredReader(cl, inv)
-	if err != nil {
-		return nil, fmt.Errorf("nns reader instantiation: %w", err)
+	for _, endpoint := range endpoints {
+		cl, err := rpcClient(ctx, endpoint)
+		if err != nil {
+			return nil, fmt.Errorf("rpcclient: %w", err)
+		}
+
+		inv := invoker.New(cl, nil)
+		nnsReader, err := nns.NewInferredReader(cl, inv)
+		if err != nil {
+			return nil, fmt.Errorf("nns readers instantiation: %w", err)
+		}
+
+		readers = append(readers, nnsReader)
 	}
 
-	return NewNNSResolver(nnsReader), nil
+	return NewNNSResolver(readers), nil
 }
 
 func rpcClient(ctx context.Context, endpoint string) (*rpcclient.Client, error) {

api/resolver/resolver.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 
 	rpcNNS "github.com/nspcc-dev/neofs-contract/rpc/nns"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
@@ -18,19 +19,21 @@ var (
 
 // NNSResolver allows to resolve container id by its name.
 type NNSResolver struct {
-	reader *rpcNNS.ContractReader
+	mu      *sync.Mutex
+	next    uint32
+	readers []*rpcNNS.ContractReader
 }
 
 // NewNNSResolver is a constructor for the NNSResolver.
-func NewNNSResolver(reader *rpcNNS.ContractReader) *NNSResolver {
-	return &NNSResolver{reader: reader}
+func NewNNSResolver(readers []*rpcNNS.ContractReader) *NNSResolver {
+	return &NNSResolver{readers: readers, mu: &sync.Mutex{}}
 }
 
 // ResolveCID looks up the container id by its name via NNS contract.
 func (r *NNSResolver) ResolveCID(_ context.Context, name string) (cid.ID, error) {
 	var result cid.ID
 
-	items, err := r.reader.GetRecords(nnsContainerDomain(name), rpcNNS.TXT)
+	items, err := r.reader().GetRecords(nnsContainerDomain(name), rpcNNS.TXT)
 	if err != nil {
 		return result, fmt.Errorf("nns get: %w", err)
 	}
@@ -46,6 +49,25 @@ func (r *NNSResolver) ResolveCID(_ context.Context, name string) (cid.ID, error)
 	return result, nil
 }
 
+func (r *NNSResolver) index() int {
+	r.mu.Lock()
+
+	r.next++
+	index := (int(r.next) - 1) % len(r.readers)
+
+	if int(r.next) >= len(r.readers) {
+		r.next = 0
+	}
+
+	r.mu.Unlock()
+
+	return index
+}
+
+func (r *NNSResolver) reader() *rpcNNS.ContractReader {
+	return r.readers[r.index()]
+}
+
 func nnsContainerDomain(name string) string {
 	return fmt.Sprintf("%s.%s", name, defaultZone)
 }

api/resolver/resolver_test.go

@@ -0,0 +1,17 @@
+package resolver
+
+import (
+	"testing"
+
+	rpcNNS "github.com/nspcc-dev/neofs-contract/rpc/nns"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_nnsResolver_index(t *testing.T) {
+	r := NewNNSResolver([]*rpcNNS.ContractReader{nil, nil, nil})
+	indexes := []int{0, 1, 2, 0, 1, 2}
+
+	for _, index := range indexes {
+		require.Equal(t, index, r.index())
+	}
+}

cmd/s3-gw/app.go

@@ -223,11 +223,11 @@ func (a *App) initMetrics() {
 }
 
 func (a *App) initResolver(ctx context.Context) {
-	endpoint := a.cfg.GetString(cfgRPCEndpoint)
+	endpoints := a.cfg.GetStringSlice(cfgRPCEndpoints)
 
-	a.log.Info("rpc endpoint", zap.String("address", endpoint))
+	a.log.Info("rpc endpoints", zap.Strings("address", endpoints))
 
-	res, err := resolver.NewContainer(ctx, endpoint)
+	res, err := resolver.NewContainer(ctx, endpoints)
 	if err != nil {
 		a.log.Fatal("resolver", zap.Error(err))
 	}
@@ -496,7 +496,7 @@ func (a *App) configReload(ctx context.Context) {
 		return
 	}
 
-	if err := a.resolverContainer.UpdateResolvers(ctx, a.cfg.GetString(cfgRPCEndpoint)); err != nil {
+	if err := a.resolverContainer.UpdateResolvers(ctx, a.cfg.GetStringSlice(cfgRPCEndpoints)); err != nil {
 		a.log.Warn("failed to update resolvers", zap.Error(err))
 	}
 

cmd/s3-gw/app_settings.go

@@ -106,7 +106,7 @@ const ( // Settings.
 	cfgTreeServiceEndpoint = "tree.service"
 
 	// NeoGo.
-	cfgRPCEndpoint = "rpc_endpoint"
+	cfgRPCEndpoints = "fschain.endpoints"
 
 	// Application.
 	cfgApplicationBuildTime = "app.build_time"
@@ -237,7 +237,7 @@ func newSettings() *viper.Viper {
 
 	peers := flags.StringArrayP(cfgPeers, "p", nil, "set NeoFS nodes")
 
-	flags.StringP(cfgRPCEndpoint, "r", "", "set RPC endpoint")
+	flags.StringP(cfgRPCEndpoints, "r", "", "set RPC endpoints")
 
 	domains := flags.StringSliceP(cfgListenDomains, "d", nil, "set domains to be listened")
 
@@ -356,7 +356,7 @@ func bindFlags(v *viper.Viper, flags *pflag.FlagSet) error {
 	if err := v.BindPFlag(cfgMaxClientsDeadline, flags.Lookup(cfgMaxClientsDeadline)); err != nil {
 		return err
 	}
-	if err := v.BindPFlag(cfgRPCEndpoint, flags.Lookup(cfgRPCEndpoint)); err != nil {
+	if err := v.BindPFlag(cfgRPCEndpoints, flags.Lookup(cfgRPCEndpoints)); err != nil {
 		return err
 	}
 

config/config.yaml

@@ -49,7 +49,9 @@ tree:
   service: node1.neofs:8080
 
 # RPC endpoint and order of resolving of bucket names
-rpc_endpoint: http://morph-chain.neofs.devenv:30333
+fschain:
+  endpoints:
+    - http://morph-chain.neofs.devenv:30333
 
 # Metrics
 pprof: