Skip to content

Commit

Permalink
agent/consul: make Client/Server config reloading more obvious
Browse files Browse the repository at this point in the history
I believe this commit also fixes a bug. Previously RPCMaxConnsPerClient was not being re-read from the RuntimeConfig, so passing it to Server.ReloadConfig was never changing the value.

Also improve the test runtime by not doing a lot of unnecessary work.
  • Loading branch information
dnephin committed Sep 28, 2020
1 parent fdc4fe8 commit 1cb3b9c
Show file tree
Hide file tree
Showing 8 changed files with 57 additions and 58 deletions.
2 changes: 1 addition & 1 deletion agent/acl_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ func (a *TestACLAgent) Shutdown() error {
func (a *TestACLAgent) Stats() map[string]map[string]string {
return nil
}
func (a *TestACLAgent) ReloadConfig(config *consul.Config) error {
func (a *TestACLAgent) ReloadConfig(_ consul.ReloadableConfig) error {
return fmt.Errorf("Unimplemented")
}

Expand Down
27 changes: 7 additions & 20 deletions agent/agent.go
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ type delegate interface {
SnapshotRPC(args *structs.SnapshotRequest, in io.Reader, out io.Writer, replyFn structs.SnapshotReplyFn) error
Shutdown() error
Stats() map[string]map[string]string
ReloadConfig(config *consul.Config) error
ReloadConfig(config consul.ReloadableConfig) error
enterpriseDelegate
}

Expand Down Expand Up @@ -3447,11 +3447,6 @@ func (a *Agent) DisableNodeMaintenance() {
a.logger.Info("Node left maintenance mode")
}

func (a *Agent) loadLimits(conf *config.RuntimeConfig) {
a.config.RPCRateLimit = conf.RPCRateLimit
a.config.RPCMaxBurst = conf.RPCMaxBurst
}

// ReloadConfig will atomically reload all configuration, including
// all services, checks, tokens, metadata, dnsServer configs, etc.
// It will also reload all ongoing watches.
Expand Down Expand Up @@ -3526,8 +3521,6 @@ func (a *Agent) reloadConfigInternal(newCfg *config.RuntimeConfig) error {
return fmt.Errorf("Failed reloading watches: %v", err)
}

a.loadLimits(newCfg)

a.httpConnLimiter.SetConfig(connlimit.Config{
MaxConnsPerClientIP: newCfg.HTTPMaxConnsPerClient,
})
Expand All @@ -3538,24 +3531,18 @@ func (a *Agent) reloadConfigInternal(newCfg *config.RuntimeConfig) error {
}
}

// this only gets used by the consulConfig function and since
// that is only ever done during init and reload here then
// an in place modification is safe as reloads cannot be
// concurrent due to both gaining a full lock on the stateLock
a.config.ConfigEntryBootstrap = newCfg.ConfigEntryBootstrap

err := a.reloadEnterprise(newCfg)
if err != nil {
return err
}

// create the config for the rpc server/client
consulCfg, err := newConsulConfig(a.config, a.logger)
if err != nil {
return err
cc := consul.ReloadableConfig{
RPCRateLimit: newCfg.RPCRateLimit,
RPCMaxBurst: newCfg.RPCMaxBurst,
RPCMaxConnsPerClient: newCfg.RPCMaxConnsPerClient,
ConfigEntryBootstrap: newCfg.ConfigEntryBootstrap,
}

if err := a.delegate.ReloadConfig(consulCfg); err != nil {
if err := a.delegate.ReloadConfig(cc); err != nil {
return err
}

Expand Down
4 changes: 2 additions & 2 deletions agent/consul/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -387,7 +387,7 @@ func (c *Client) GetLANCoordinate() (lib.CoordinateSet, error) {

// ReloadConfig is used to have the Client do an online reload of
// relevant configuration information
func (c *Client) ReloadConfig(config *Config) error {
c.rpcLimiter.Store(rate.NewLimiter(config.RPCRate, config.RPCMaxBurst))
func (c *Client) ReloadConfig(config ReloadableConfig) error {
c.rpcLimiter.Store(rate.NewLimiter(config.RPCRateLimit, config.RPCMaxBurst))
return nil
}
35 changes: 19 additions & 16 deletions agent/consul/client_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,12 @@ import (
"testing"
"time"

"github.com/hashicorp/go-hclog"
msgpackrpc "github.com/hashicorp/net-rpc-msgpackrpc"
"github.com/hashicorp/serf/serf"
"github.com/stretchr/testify/require"
"golang.org/x/time/rate"

"github.com/hashicorp/consul/agent/pool"
"github.com/hashicorp/consul/agent/router"
"github.com/hashicorp/consul/agent/structs"
Expand All @@ -18,11 +24,6 @@ import (
"github.com/hashicorp/consul/sdk/testutil/retry"
"github.com/hashicorp/consul/testrpc"
"github.com/hashicorp/consul/tlsutil"
"github.com/hashicorp/go-hclog"
msgpackrpc "github.com/hashicorp/net-rpc-msgpackrpc"
"github.com/hashicorp/serf/serf"
"github.com/stretchr/testify/require"
"golang.org/x/time/rate"
)

func testClientConfig(t *testing.T) (string, *Config) {
Expand Down Expand Up @@ -726,23 +727,25 @@ func TestClientServer_UserEvent(t *testing.T) {
}
}

func TestClient_Reload(t *testing.T) {
t.Parallel()
dir1, c := testClientWithConfig(t, func(c *Config) {
c.RPCRate = 500
c.RPCMaxBurst = 5000
})
defer os.RemoveAll(dir1)
defer c.Shutdown()
func TestClient_ReloadConfig(t *testing.T) {
_, cfg := testClientConfig(t)
cfg.RPCRate = rate.Limit(500)
cfg.RPCMaxBurst = 5000
deps := newDefaultDeps(t, &Config{NodeName: "node1", Datacenter: "dc1"})
c, err := NewClient(cfg, deps)
require.NoError(t, err)

limiter := c.rpcLimiter.Load().(*rate.Limiter)
require.Equal(t, rate.Limit(500), limiter.Limit())
require.Equal(t, 5000, limiter.Burst())

c.config.RPCRate = 1000
c.config.RPCMaxBurst = 10000
rc := ReloadableConfig{
RPCRateLimit: 1000,
RPCMaxBurst: 10000,
RPCMaxConnsPerClient: 0,
}
require.NoError(t, c.ReloadConfig(rc))

require.NoError(t, c.ReloadConfig(c.config))
limiter = c.rpcLimiter.Load().(*rate.Limiter)
require.Equal(t, rate.Limit(1000), limiter.Limit())
require.Equal(t, 10000, limiter.Burst())
Expand Down
9 changes: 9 additions & 0 deletions agent/consul/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -639,3 +639,12 @@ func DefaultConfig() *Config {

return conf
}

// ReloadableConfig is the configuration that is passed to ReloadConfig when
// application config is reloaded.
type ReloadableConfig struct {
RPCRateLimit rate.Limit
RPCMaxBurst int
RPCMaxConnsPerClient int
ConfigEntryBootstrap []structs.ConfigEntry
}
9 changes: 6 additions & 3 deletions agent/consul/rpc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -648,9 +648,12 @@ func TestRPC_RPCMaxConnsPerClient(t *testing.T) {
defer conn4.Close()

// Reload config with higher limit
newCfg := *s1.config
newCfg.RPCMaxConnsPerClient = 10
require.NoError(t, s1.ReloadConfig(&newCfg))
rc := ReloadableConfig{
RPCRateLimit: s1.config.RPCRate,
RPCMaxBurst: s1.config.RPCMaxBurst,
RPCMaxConnsPerClient: 10,
}
require.NoError(t, s1.ReloadConfig(rc))

// Now another conn should be allowed
conn5 := connectClient(t, s1, tc.magicByte, tc.tlsEnabled, true, "conn5")
Expand Down
4 changes: 2 additions & 2 deletions agent/consul/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -1391,8 +1391,8 @@ func (s *Server) GetLANCoordinate() (lib.CoordinateSet, error) {

// ReloadConfig is used to have the Server do an online reload of
// relevant configuration information
func (s *Server) ReloadConfig(config *Config) error {
s.rpcLimiter.Store(rate.NewLimiter(config.RPCRate, config.RPCMaxBurst))
func (s *Server) ReloadConfig(config ReloadableConfig) error {
s.rpcLimiter.Store(rate.NewLimiter(config.RPCRateLimit, config.RPCMaxBurst))
s.rpcConnLimiter.SetConfig(connlimit.Config{
MaxConnsPerClientIP: config.RPCMaxConnsPerClient,
})
Expand Down
25 changes: 11 additions & 14 deletions agent/consul/server_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1394,10 +1394,10 @@ func TestServer_RevokeLeadershipIdempotent(t *testing.T) {
s1.revokeLeadership()
}

func TestServer_Reload(t *testing.T) {
func TestServer_ReloadConfig(t *testing.T) {
t.Parallel()

global_entry_init := &structs.ProxyConfigEntry{
entryInit := &structs.ProxyConfigEntry{
Kind: structs.ProxyDefaults,
Name: structs.ProxyConfigGlobal,
Config: map[string]interface{}{
Expand All @@ -1418,28 +1418,25 @@ func TestServer_Reload(t *testing.T) {

testrpc.WaitForTestAgent(t, s.RPC, "dc1")

s.config.ConfigEntryBootstrap = []structs.ConfigEntry{
global_entry_init,
}

limiter := s.rpcLimiter.Load().(*rate.Limiter)
require.Equal(t, rate.Limit(500), limiter.Limit())
require.Equal(t, 5000, limiter.Burst())

// Change rate limit
s.config.RPCRate = 1000
s.config.RPCMaxBurst = 10000

s.ReloadConfig(s.config)
rc := ReloadableConfig{
RPCRateLimit: 1000,
RPCMaxBurst: 10000,
ConfigEntryBootstrap: []structs.ConfigEntry{entryInit},
}
require.NoError(t, s.ReloadConfig(rc))

_, entry, err := s.fsm.State().ConfigEntry(nil, structs.ProxyDefaults, structs.ProxyConfigGlobal, structs.DefaultEnterpriseMeta())
require.NoError(t, err)
require.NotNil(t, entry)
global, ok := entry.(*structs.ProxyConfigEntry)
require.True(t, ok)
require.Equal(t, global_entry_init.Kind, global.Kind)
require.Equal(t, global_entry_init.Name, global.Name)
require.Equal(t, global_entry_init.Config, global.Config)
require.Equal(t, entryInit.Kind, global.Kind)
require.Equal(t, entryInit.Name, global.Name)
require.Equal(t, entryInit.Config, global.Config)

// Check rate limiter got updated
limiter = s.rpcLimiter.Load().(*rate.Limiter)
Expand Down

0 comments on commit 1cb3b9c

Please sign in to comment.