Files
nats-server/server/reload.go
Ivan Kozlovic 2d181e1c27 [FIXED] Routing: TLS connections to discovered server may fail
The server was not setting "server name" in the TLS configuration
for route connections, which may lead to failed (re)connect if
the certificate does not allow for the IP and the URL did not
have the hostname, which would happen with gossip protocol.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2022-11-07 17:26:17 -07:00

1987 lines
58 KiB
Go

// Copyright 2017-2022 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package server
import (
"crypto/tls"
"errors"
"fmt"
"net/url"
"reflect"
"sort"
"strings"
"sync/atomic"
"time"
"github.com/nats-io/jwt/v2"
)
// FlagSnapshot captures the server options as specified by CLI flags at
// startup. This should not be modified once the server has started.
var FlagSnapshot *Options
type reloadContext struct {
oldClusterPerms *RoutePermissions
}
// option is a hot-swappable configuration setting.
type option interface {
// Apply the server option.
Apply(server *Server)
// IsLoggingChange indicates if this option requires reloading the logger.
IsLoggingChange() bool
// IsTraceLevelChange indicates if this option requires reloading cached trace level.
// Clients store trace level separately.
IsTraceLevelChange() bool
// IsAuthChange indicates if this option requires reloading authorization.
IsAuthChange() bool
// IsTLSChange indicates if this option requires reloading TLS.
IsTLSChange() bool
// IsClusterPermsChange indicates if this option requires reloading
// cluster permissions.
IsClusterPermsChange() bool
// IsJetStreamChange inidicates a change in the servers config for JetStream.
// Account changes will be handled separately in reloadAuthorization.
IsJetStreamChange() bool
// Indicates a change in the server that requires publishing the server's statz
IsStatszChange() bool
}
// noopOption is a base struct that provides default no-op behaviors.
type noopOption struct{}
func (n noopOption) IsLoggingChange() bool {
return false
}
func (n noopOption) IsTraceLevelChange() bool {
return false
}
func (n noopOption) IsAuthChange() bool {
return false
}
func (n noopOption) IsTLSChange() bool {
return false
}
func (n noopOption) IsClusterPermsChange() bool {
return false
}
func (n noopOption) IsJetStreamChange() bool {
return false
}
func (n noopOption) IsStatszChange() bool {
return false
}
// loggingOption is a base struct that provides default option behaviors for
// logging-related options.
type loggingOption struct {
noopOption
}
func (l loggingOption) IsLoggingChange() bool {
return true
}
// traceLevelOption is a base struct that provides default option behaviors for
// tracelevel-related options.
type traceLevelOption struct {
loggingOption
}
func (l traceLevelOption) IsTraceLevelChange() bool {
return true
}
// traceOption implements the option interface for the `trace` setting.
type traceOption struct {
traceLevelOption
newValue bool
}
// Apply is a no-op because logging will be reloaded after options are applied.
func (t *traceOption) Apply(server *Server) {
server.Noticef("Reloaded: trace = %v", t.newValue)
}
// traceOption implements the option interface for the `trace` setting.
type traceVerboseOption struct {
traceLevelOption
newValue bool
}
// Apply is a no-op because logging will be reloaded after options are applied.
func (t *traceVerboseOption) Apply(server *Server) {
server.Noticef("Reloaded: trace_verbose = %v", t.newValue)
}
// debugOption implements the option interface for the `debug` setting.
type debugOption struct {
loggingOption
newValue bool
}
// Apply is mostly a no-op because logging will be reloaded after options are applied.
// However we will kick the raft nodes if they exist to reload.
func (d *debugOption) Apply(server *Server) {
server.Noticef("Reloaded: debug = %v", d.newValue)
server.reloadDebugRaftNodes()
}
// logtimeOption implements the option interface for the `logtime` setting.
type logtimeOption struct {
loggingOption
newValue bool
}
// Apply is a no-op because logging will be reloaded after options are applied.
func (l *logtimeOption) Apply(server *Server) {
server.Noticef("Reloaded: logtime = %v", l.newValue)
}
// logfileOption implements the option interface for the `log_file` setting.
type logfileOption struct {
loggingOption
newValue string
}
// Apply is a no-op because logging will be reloaded after options are applied.
func (l *logfileOption) Apply(server *Server) {
server.Noticef("Reloaded: log_file = %v", l.newValue)
}
// syslogOption implements the option interface for the `syslog` setting.
type syslogOption struct {
loggingOption
newValue bool
}
// Apply is a no-op because logging will be reloaded after options are applied.
func (s *syslogOption) Apply(server *Server) {
server.Noticef("Reloaded: syslog = %v", s.newValue)
}
// remoteSyslogOption implements the option interface for the `remote_syslog`
// setting.
type remoteSyslogOption struct {
loggingOption
newValue string
}
// Apply is a no-op because logging will be reloaded after options are applied.
func (r *remoteSyslogOption) Apply(server *Server) {
server.Noticef("Reloaded: remote_syslog = %v", r.newValue)
}
// tlsOption implements the option interface for the `tls` setting.
type tlsOption struct {
noopOption
newValue *tls.Config
}
// Apply the tls change.
func (t *tlsOption) Apply(server *Server) {
server.mu.Lock()
tlsRequired := t.newValue != nil
server.info.TLSRequired = tlsRequired && !server.getOpts().AllowNonTLS
message := "disabled"
if tlsRequired {
server.info.TLSVerify = (t.newValue.ClientAuth == tls.RequireAndVerifyClientCert)
message = "enabled"
}
server.mu.Unlock()
server.Noticef("Reloaded: tls = %s", message)
}
func (t *tlsOption) IsTLSChange() bool {
return true
}
// tlsTimeoutOption implements the option interface for the tls `timeout`
// setting.
type tlsTimeoutOption struct {
noopOption
newValue float64
}
// Apply is a no-op because the timeout will be reloaded after options are
// applied.
func (t *tlsTimeoutOption) Apply(server *Server) {
server.Noticef("Reloaded: tls timeout = %v", t.newValue)
}
// tlsPinnedCertOption implements the option interface for the tls `pinned_certs` setting.
type tlsPinnedCertOption struct {
noopOption
newValue PinnedCertSet
}
// Apply is a no-op because the pinned certs will be reloaded after options are applied.
func (t *tlsPinnedCertOption) Apply(server *Server) {
server.Noticef("Reloaded: %d pinned_certs", len(t.newValue))
}
// authOption is a base struct that provides default option behaviors.
type authOption struct {
noopOption
}
func (o authOption) IsAuthChange() bool {
return true
}
// usernameOption implements the option interface for the `username` setting.
type usernameOption struct {
authOption
}
// Apply is a no-op because authorization will be reloaded after options are
// applied.
func (u *usernameOption) Apply(server *Server) {
server.Noticef("Reloaded: authorization username")
}
// passwordOption implements the option interface for the `password` setting.
type passwordOption struct {
authOption
}
// Apply is a no-op because authorization will be reloaded after options are
// applied.
func (p *passwordOption) Apply(server *Server) {
server.Noticef("Reloaded: authorization password")
}
// authorizationOption implements the option interface for the `token`
// authorization setting.
type authorizationOption struct {
authOption
}
// Apply is a no-op because authorization will be reloaded after options are
// applied.
func (a *authorizationOption) Apply(server *Server) {
server.Noticef("Reloaded: authorization token")
}
// authTimeoutOption implements the option interface for the authorization
// `timeout` setting.
type authTimeoutOption struct {
noopOption // Not authOption because this is a no-op; will be reloaded with options.
newValue float64
}
// Apply is a no-op because the timeout will be reloaded after options are
// applied.
func (a *authTimeoutOption) Apply(server *Server) {
server.Noticef("Reloaded: authorization timeout = %v", a.newValue)
}
// tagsOption implements the option interface for the `tags` setting.
type tagsOption struct {
noopOption // Not authOption because this is a no-op; will be reloaded with options.
}
func (u *tagsOption) Apply(server *Server) {
server.Noticef("Reloaded: tags")
}
func (u *tagsOption) IsStatszChange() bool {
return true
}
// usersOption implements the option interface for the authorization `users`
// setting.
type usersOption struct {
authOption
}
func (u *usersOption) Apply(server *Server) {
server.Noticef("Reloaded: authorization users")
}
// nkeysOption implements the option interface for the authorization `users`
// setting.
type nkeysOption struct {
authOption
}
func (u *nkeysOption) Apply(server *Server) {
server.Noticef("Reloaded: authorization nkey users")
}
// clusterOption implements the option interface for the `cluster` setting.
type clusterOption struct {
authOption
newValue ClusterOpts
permsChanged bool
}
// Apply the cluster change.
func (c *clusterOption) Apply(s *Server) {
// TODO: support enabling/disabling clustering.
s.mu.Lock()
tlsRequired := c.newValue.TLSConfig != nil
s.routeInfo.TLSRequired = tlsRequired
s.routeInfo.TLSVerify = tlsRequired
s.routeInfo.AuthRequired = c.newValue.Username != ""
if c.newValue.NoAdvertise {
s.routeInfo.ClientConnectURLs = nil
s.routeInfo.WSConnectURLs = nil
} else {
s.routeInfo.ClientConnectURLs = s.clientConnectURLs
s.routeInfo.WSConnectURLs = s.websocket.connectURLs
}
s.setRouteInfoHostPortAndIP()
s.mu.Unlock()
if c.newValue.Name != "" && c.newValue.Name != s.ClusterName() {
s.setClusterName(c.newValue.Name)
}
s.Noticef("Reloaded: cluster")
if tlsRequired && c.newValue.TLSConfig.InsecureSkipVerify {
s.Warnf(clusterTLSInsecureWarning)
}
}
func (c *clusterOption) IsClusterPermsChange() bool {
return c.permsChanged
}
// routesOption implements the option interface for the cluster `routes`
// setting.
type routesOption struct {
noopOption
add []*url.URL
remove []*url.URL
}
// Apply the route changes by adding and removing the necessary routes.
func (r *routesOption) Apply(server *Server) {
server.mu.Lock()
routes := make([]*client, len(server.routes))
i := 0
for _, client := range server.routes {
routes[i] = client
i++
}
// If there was a change, notify monitoring code that it should
// update the route URLs if /varz endpoint is inspected.
if len(r.add)+len(r.remove) > 0 {
server.varzUpdateRouteURLs = true
}
server.mu.Unlock()
// Remove routes.
for _, remove := range r.remove {
for _, client := range routes {
var url *url.URL
client.mu.Lock()
if client.route != nil {
url = client.route.url
}
client.mu.Unlock()
if url != nil && urlsAreEqual(url, remove) {
// Do not attempt to reconnect when route is removed.
client.setNoReconnect()
client.closeConnection(RouteRemoved)
server.Noticef("Removed route %v", remove)
}
}
}
// Add routes.
server.mu.Lock()
server.solicitRoutes(r.add)
server.mu.Unlock()
server.Noticef("Reloaded: cluster routes")
}
// maxConnOption implements the option interface for the `max_connections`
// setting.
type maxConnOption struct {
noopOption
newValue int
}
// Apply the max connections change by closing random connections til we are
// below the limit if necessary.
func (m *maxConnOption) Apply(server *Server) {
server.mu.Lock()
var (
clients = make([]*client, len(server.clients))
i = 0
)
// Map iteration is random, which allows us to close random connections.
for _, client := range server.clients {
clients[i] = client
i++
}
server.mu.Unlock()
if m.newValue > 0 && len(clients) > m.newValue {
// Close connections til we are within the limit.
var (
numClose = len(clients) - m.newValue
closed = 0
)
for _, client := range clients {
client.maxConnExceeded()
closed++
if closed >= numClose {
break
}
}
server.Noticef("Closed %d connections to fall within max_connections", closed)
}
server.Noticef("Reloaded: max_connections = %v", m.newValue)
}
// pidFileOption implements the option interface for the `pid_file` setting.
type pidFileOption struct {
noopOption
newValue string
}
// Apply the setting by logging the pid to the new file.
func (p *pidFileOption) Apply(server *Server) {
if p.newValue == "" {
return
}
if err := server.logPid(); err != nil {
server.Errorf("Failed to write pidfile: %v", err)
}
server.Noticef("Reloaded: pid_file = %v", p.newValue)
}
// portsFileDirOption implements the option interface for the `portFileDir` setting.
type portsFileDirOption struct {
noopOption
oldValue string
newValue string
}
func (p *portsFileDirOption) Apply(server *Server) {
server.deletePortsFile(p.oldValue)
server.logPorts()
server.Noticef("Reloaded: ports_file_dir = %v", p.newValue)
}
// maxControlLineOption implements the option interface for the
// `max_control_line` setting.
type maxControlLineOption struct {
noopOption
newValue int32
}
// Apply the setting by updating each client.
func (m *maxControlLineOption) Apply(server *Server) {
mcl := int32(m.newValue)
server.mu.Lock()
for _, client := range server.clients {
atomic.StoreInt32(&client.mcl, mcl)
}
server.mu.Unlock()
server.Noticef("Reloaded: max_control_line = %d", mcl)
}
// maxPayloadOption implements the option interface for the `max_payload`
// setting.
type maxPayloadOption struct {
noopOption
newValue int32
}
// Apply the setting by updating the server info and each client.
func (m *maxPayloadOption) Apply(server *Server) {
server.mu.Lock()
server.info.MaxPayload = m.newValue
for _, client := range server.clients {
atomic.StoreInt32(&client.mpay, int32(m.newValue))
}
server.mu.Unlock()
server.Noticef("Reloaded: max_payload = %d", m.newValue)
}
// pingIntervalOption implements the option interface for the `ping_interval`
// setting.
type pingIntervalOption struct {
noopOption
newValue time.Duration
}
// Apply is a no-op because the ping interval will be reloaded after options
// are applied.
func (p *pingIntervalOption) Apply(server *Server) {
server.Noticef("Reloaded: ping_interval = %s", p.newValue)
}
// maxPingsOutOption implements the option interface for the `ping_max`
// setting.
type maxPingsOutOption struct {
noopOption
newValue int
}
// Apply is a no-op because the ping interval will be reloaded after options
// are applied.
func (m *maxPingsOutOption) Apply(server *Server) {
server.Noticef("Reloaded: ping_max = %d", m.newValue)
}
// writeDeadlineOption implements the option interface for the `write_deadline`
// setting.
type writeDeadlineOption struct {
noopOption
newValue time.Duration
}
// Apply is a no-op because the write deadline will be reloaded after options
// are applied.
func (w *writeDeadlineOption) Apply(server *Server) {
server.Noticef("Reloaded: write_deadline = %s", w.newValue)
}
// clientAdvertiseOption implements the option interface for the `client_advertise` setting.
type clientAdvertiseOption struct {
noopOption
newValue string
}
// Apply the setting by updating the server info and regenerate the infoJSON byte array.
func (c *clientAdvertiseOption) Apply(server *Server) {
server.mu.Lock()
server.setInfoHostPort()
server.mu.Unlock()
server.Noticef("Reload: client_advertise = %s", c.newValue)
}
// accountsOption implements the option interface.
// Ensure that authorization code is executed if any change in accounts
type accountsOption struct {
authOption
}
// Apply is a no-op. Changes will be applied in reloadAuthorization
func (a *accountsOption) Apply(s *Server) {
s.Noticef("Reloaded: accounts")
}
// For changes to a server's config.
type jetStreamOption struct {
noopOption
newValue bool
}
func (a *jetStreamOption) Apply(s *Server) {
s.Noticef("Reloaded: JetStream")
}
func (jso jetStreamOption) IsJetStreamChange() bool {
return true
}
func (jso jetStreamOption) IsStatszChange() bool {
return true
}
type ocspOption struct {
noopOption
newValue *OCSPConfig
}
func (a *ocspOption) Apply(s *Server) {
s.Noticef("Reloaded: OCSP")
}
// connectErrorReports implements the option interface for the `connect_error_reports`
// setting.
type connectErrorReports struct {
noopOption
newValue int
}
// Apply is a no-op because the value will be reloaded after options are applied.
func (c *connectErrorReports) Apply(s *Server) {
s.Noticef("Reloaded: connect_error_reports = %v", c.newValue)
}
// connectErrorReports implements the option interface for the `connect_error_reports`
// setting.
type reconnectErrorReports struct {
noopOption
newValue int
}
// Apply is a no-op because the value will be reloaded after options are applied.
func (r *reconnectErrorReports) Apply(s *Server) {
s.Noticef("Reloaded: reconnect_error_reports = %v", r.newValue)
}
// maxTracedMsgLenOption implements the option interface for the `max_traced_msg_len` setting.
type maxTracedMsgLenOption struct {
noopOption
newValue int
}
// Apply the setting by updating the maximum traced message length.
func (m *maxTracedMsgLenOption) Apply(server *Server) {
server.mu.Lock()
defer server.mu.Unlock()
server.opts.MaxTracedMsgLen = m.newValue
server.Noticef("Reloaded: max_traced_msg_len = %d", m.newValue)
}
type mqttAckWaitReload struct {
noopOption
newValue time.Duration
}
func (o *mqttAckWaitReload) Apply(s *Server) {
s.Noticef("Reloaded: MQTT ack_wait = %v", o.newValue)
}
type mqttMaxAckPendingReload struct {
noopOption
newValue uint16
}
func (o *mqttMaxAckPendingReload) Apply(s *Server) {
s.mqttUpdateMaxAckPending(o.newValue)
s.Noticef("Reloaded: MQTT max_ack_pending = %v", o.newValue)
}
type mqttStreamReplicasReload struct {
noopOption
newValue int
}
func (o *mqttStreamReplicasReload) Apply(s *Server) {
s.Noticef("Reloaded: MQTT stream_replicas = %v", o.newValue)
}
type mqttConsumerReplicasReload struct {
noopOption
newValue int
}
func (o *mqttConsumerReplicasReload) Apply(s *Server) {
s.Noticef("Reloaded: MQTT consumer_replicas = %v", o.newValue)
}
type mqttConsumerMemoryStorageReload struct {
noopOption
newValue bool
}
func (o *mqttConsumerMemoryStorageReload) Apply(s *Server) {
s.Noticef("Reloaded: MQTT consumer_memory_storage = %v", o.newValue)
}
type mqttInactiveThresholdReload struct {
noopOption
newValue time.Duration
}
func (o *mqttInactiveThresholdReload) Apply(s *Server) {
s.Noticef("Reloaded: MQTT consumer_inactive_threshold = %v", o.newValue)
}
// Compares options and disconnects clients that are no longer listed in pinned certs. Lock must not be held.
func (s *Server) recheckPinnedCerts(curOpts *Options, newOpts *Options) {
s.mu.Lock()
disconnectClients := []*client{}
protoToPinned := map[int]PinnedCertSet{}
if !reflect.DeepEqual(newOpts.TLSPinnedCerts, curOpts.TLSPinnedCerts) {
protoToPinned[NATS] = curOpts.TLSPinnedCerts
}
if !reflect.DeepEqual(newOpts.MQTT.TLSPinnedCerts, curOpts.MQTT.TLSPinnedCerts) {
protoToPinned[MQTT] = curOpts.MQTT.TLSPinnedCerts
}
if !reflect.DeepEqual(newOpts.Websocket.TLSPinnedCerts, curOpts.Websocket.TLSPinnedCerts) {
protoToPinned[WS] = curOpts.Websocket.TLSPinnedCerts
}
for _, c := range s.clients {
if c.kind != CLIENT {
continue
}
if pinned, ok := protoToPinned[c.clientType()]; ok {
if !c.matchesPinnedCert(pinned) {
disconnectClients = append(disconnectClients, c)
}
}
}
checkClients := func(kind int, clients map[uint64]*client, set PinnedCertSet) {
for _, c := range clients {
if c.kind == kind && !c.matchesPinnedCert(set) {
disconnectClients = append(disconnectClients, c)
}
}
}
if !reflect.DeepEqual(newOpts.LeafNode.TLSPinnedCerts, curOpts.LeafNode.TLSPinnedCerts) {
checkClients(LEAF, s.leafs, newOpts.LeafNode.TLSPinnedCerts)
}
if !reflect.DeepEqual(newOpts.Cluster.TLSPinnedCerts, curOpts.Cluster.TLSPinnedCerts) {
checkClients(ROUTER, s.routes, newOpts.Cluster.TLSPinnedCerts)
}
if reflect.DeepEqual(newOpts.Gateway.TLSPinnedCerts, curOpts.Gateway.TLSPinnedCerts) {
for _, c := range s.remotes {
if !c.matchesPinnedCert(newOpts.Gateway.TLSPinnedCerts) {
disconnectClients = append(disconnectClients, c)
}
}
}
s.mu.Unlock()
if len(disconnectClients) > 0 {
s.Noticef("Disconnect %d clients due to pinned certs reload", len(disconnectClients))
for _, c := range disconnectClients {
c.closeConnection(TLSHandshakeError)
}
}
}
// Reload reads the current configuration file and calls out to ReloadOptions
// to apply the changes. This returns an error if the server was not started
// with a config file or an option which doesn't support hot-swapping was changed.
func (s *Server) Reload() error {
s.mu.Lock()
configFile := s.configFile
s.mu.Unlock()
if configFile == "" {
return errors.New("can only reload config when a file is provided using -c or --config")
}
newOpts, err := ProcessConfigFile(configFile)
if err != nil {
// TODO: Dump previous good config to a .bak file?
return err
}
return s.ReloadOptions(newOpts)
}
// ReloadOptions applies any supported options from the provided Option
// type. This returns an error if an option which doesn't support
// hot-swapping was changed.
func (s *Server) ReloadOptions(newOpts *Options) error {
s.mu.Lock()
s.reloading = true
defer func() {
s.mu.Lock()
s.reloading = false
s.mu.Unlock()
}()
curOpts := s.getOpts()
// Wipe trusted keys if needed when we have an operator.
if len(curOpts.TrustedOperators) > 0 && len(curOpts.TrustedKeys) > 0 {
curOpts.TrustedKeys = nil
}
clientOrgPort := curOpts.Port
clusterOrgPort := curOpts.Cluster.Port
gatewayOrgPort := curOpts.Gateway.Port
leafnodesOrgPort := curOpts.LeafNode.Port
websocketOrgPort := curOpts.Websocket.Port
mqttOrgPort := curOpts.MQTT.Port
s.mu.Unlock()
// Apply flags over config file settings.
newOpts = MergeOptions(newOpts, FlagSnapshot)
// Need more processing for boolean flags...
if FlagSnapshot != nil {
applyBoolFlags(newOpts, FlagSnapshot)
}
setBaselineOptions(newOpts)
// setBaselineOptions sets Port to 0 if set to -1 (RANDOM port)
// If that's the case, set it to the saved value when the accept loop was
// created.
if newOpts.Port == 0 {
newOpts.Port = clientOrgPort
}
// We don't do that for cluster, so check against -1.
if newOpts.Cluster.Port == -1 {
newOpts.Cluster.Port = clusterOrgPort
}
if newOpts.Gateway.Port == -1 {
newOpts.Gateway.Port = gatewayOrgPort
}
if newOpts.LeafNode.Port == -1 {
newOpts.LeafNode.Port = leafnodesOrgPort
}
if newOpts.Websocket.Port == -1 {
newOpts.Websocket.Port = websocketOrgPort
}
if newOpts.MQTT.Port == -1 {
newOpts.MQTT.Port = mqttOrgPort
}
if err := s.reloadOptions(curOpts, newOpts); err != nil {
return err
}
s.recheckPinnedCerts(curOpts, newOpts)
s.mu.Lock()
s.configTime = time.Now().UTC()
s.updateVarzConfigReloadableFields(s.varz)
s.mu.Unlock()
return nil
}
func applyBoolFlags(newOpts, flagOpts *Options) {
// Reset fields that may have been set to `true` in
// MergeOptions() when some of the flags default to `true`
// but have not been explicitly set and therefore value
// from config file should take precedence.
for name, val := range newOpts.inConfig {
f := reflect.ValueOf(newOpts).Elem()
names := strings.Split(name, ".")
for _, name := range names {
f = f.FieldByName(name)
}
f.SetBool(val)
}
// Now apply value (true or false) from flags that have
// been explicitly set in command line
for name, val := range flagOpts.inCmdLine {
f := reflect.ValueOf(newOpts).Elem()
names := strings.Split(name, ".")
for _, name := range names {
f = f.FieldByName(name)
}
f.SetBool(val)
}
}
// reloadOptions reloads the server config with the provided options. If an
// option that doesn't support hot-swapping is changed, this returns an error.
func (s *Server) reloadOptions(curOpts, newOpts *Options) error {
// Apply to the new options some of the options that may have been set
// that can't be configured in the config file (this can happen in
// applications starting NATS Server programmatically).
newOpts.CustomClientAuthentication = curOpts.CustomClientAuthentication
newOpts.CustomRouterAuthentication = curOpts.CustomRouterAuthentication
changed, err := s.diffOptions(newOpts)
if err != nil {
return err
}
if len(changed) != 0 {
if err := validateOptions(newOpts); err != nil {
return err
}
}
// Create a context that is used to pass special info that we may need
// while applying the new options.
ctx := reloadContext{oldClusterPerms: curOpts.Cluster.Permissions}
s.setOpts(newOpts)
s.applyOptions(&ctx, changed)
return nil
}
// For the purpose of comparing, impose a order on slice data types where order does not matter
func imposeOrder(value interface{}) error {
switch value := value.(type) {
case []*Account:
sort.Slice(value, func(i, j int) bool {
return value[i].Name < value[j].Name
})
for _, a := range value {
sort.Slice(a.imports.streams, func(i, j int) bool {
return a.imports.streams[i].acc.Name < a.imports.streams[j].acc.Name
})
}
case []*User:
sort.Slice(value, func(i, j int) bool {
return value[i].Username < value[j].Username
})
case []*NkeyUser:
sort.Slice(value, func(i, j int) bool {
return value[i].Nkey < value[j].Nkey
})
case []*url.URL:
sort.Slice(value, func(i, j int) bool {
return value[i].String() < value[j].String()
})
case []string:
sort.Strings(value)
case []*jwt.OperatorClaims:
sort.Slice(value, func(i, j int) bool {
return value[i].Issuer < value[j].Issuer
})
case GatewayOpts:
sort.Slice(value.Gateways, func(i, j int) bool {
return value.Gateways[i].Name < value.Gateways[j].Name
})
case WebsocketOpts:
sort.Strings(value.AllowedOrigins)
case string, bool, uint8, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config, PinnedCertSet,
*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList,
*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher:
// explicitly skipped types
default:
// this will fail during unit tests
return fmt.Errorf("OnReload, sort or explicitly skip type: %s",
reflect.TypeOf(value))
}
return nil
}
// diffOptions returns a slice containing options which have been changed. If
// an option that doesn't support hot-swapping is changed, this returns an
// error.
func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
var (
oldConfig = reflect.ValueOf(s.getOpts()).Elem()
newConfig = reflect.ValueOf(newOpts).Elem()
diffOpts = []option{}
// Need to keep track of whether JS is being disabled
// to prevent changing limits at runtime.
jsEnabled = s.JetStreamEnabled()
disableJS bool
jsMemLimitsChanged bool
jsFileLimitsChanged bool
jsStoreDirChanged bool
)
for i := 0; i < oldConfig.NumField(); i++ {
field := oldConfig.Type().Field(i)
// field.PkgPath is empty for exported fields, and is not for unexported ones.
// We skip the unexported fields.
if field.PkgPath != _EMPTY_ {
continue
}
var (
oldValue = oldConfig.Field(i).Interface()
newValue = newConfig.Field(i).Interface()
)
if err := imposeOrder(oldValue); err != nil {
return nil, err
}
if err := imposeOrder(newValue); err != nil {
return nil, err
}
optName := strings.ToLower(field.Name)
// accounts and users (referencing accounts) will always differ as accounts
// contain internal state, say locks etc..., so we don't bother here.
// This also avoids races with atomic stats counters
if optName != "accounts" && optName != "users" {
if changed := !reflect.DeepEqual(oldValue, newValue); !changed {
// Check to make sure we are running JetStream if we think we should be.
if optName == "jetstream" && newValue.(bool) {
if !jsEnabled {
diffOpts = append(diffOpts, &jetStreamOption{newValue: true})
}
}
continue
}
}
switch optName {
case "traceverbose":
diffOpts = append(diffOpts, &traceVerboseOption{newValue: newValue.(bool)})
case "trace":
diffOpts = append(diffOpts, &traceOption{newValue: newValue.(bool)})
case "debug":
diffOpts = append(diffOpts, &debugOption{newValue: newValue.(bool)})
case "logtime":
diffOpts = append(diffOpts, &logtimeOption{newValue: newValue.(bool)})
case "logfile":
diffOpts = append(diffOpts, &logfileOption{newValue: newValue.(string)})
case "syslog":
diffOpts = append(diffOpts, &syslogOption{newValue: newValue.(bool)})
case "remotesyslog":
diffOpts = append(diffOpts, &remoteSyslogOption{newValue: newValue.(string)})
case "tlsconfig":
diffOpts = append(diffOpts, &tlsOption{newValue: newValue.(*tls.Config)})
case "tlstimeout":
diffOpts = append(diffOpts, &tlsTimeoutOption{newValue: newValue.(float64)})
case "tlspinnedcerts":
diffOpts = append(diffOpts, &tlsPinnedCertOption{newValue: newValue.(PinnedCertSet)})
case "username":
diffOpts = append(diffOpts, &usernameOption{})
case "password":
diffOpts = append(diffOpts, &passwordOption{})
case "tags":
diffOpts = append(diffOpts, &tagsOption{})
case "authorization":
diffOpts = append(diffOpts, &authorizationOption{})
case "authtimeout":
diffOpts = append(diffOpts, &authTimeoutOption{newValue: newValue.(float64)})
case "users":
diffOpts = append(diffOpts, &usersOption{})
case "nkeys":
diffOpts = append(diffOpts, &nkeysOption{})
case "cluster":
newClusterOpts := newValue.(ClusterOpts)
oldClusterOpts := oldValue.(ClusterOpts)
if err := validateClusterOpts(oldClusterOpts, newClusterOpts); err != nil {
return nil, err
}
permsChanged := !reflect.DeepEqual(newClusterOpts.Permissions, oldClusterOpts.Permissions)
diffOpts = append(diffOpts, &clusterOption{newValue: newClusterOpts, permsChanged: permsChanged})
case "routes":
add, remove := diffRoutes(oldValue.([]*url.URL), newValue.([]*url.URL))
diffOpts = append(diffOpts, &routesOption{add: add, remove: remove})
case "maxconn":
diffOpts = append(diffOpts, &maxConnOption{newValue: newValue.(int)})
case "pidfile":
diffOpts = append(diffOpts, &pidFileOption{newValue: newValue.(string)})
case "portsfiledir":
diffOpts = append(diffOpts, &portsFileDirOption{newValue: newValue.(string), oldValue: oldValue.(string)})
case "maxcontrolline":
diffOpts = append(diffOpts, &maxControlLineOption{newValue: newValue.(int32)})
case "maxpayload":
diffOpts = append(diffOpts, &maxPayloadOption{newValue: newValue.(int32)})
case "pinginterval":
diffOpts = append(diffOpts, &pingIntervalOption{newValue: newValue.(time.Duration)})
case "maxpingsout":
diffOpts = append(diffOpts, &maxPingsOutOption{newValue: newValue.(int)})
case "writedeadline":
diffOpts = append(diffOpts, &writeDeadlineOption{newValue: newValue.(time.Duration)})
case "clientadvertise":
cliAdv := newValue.(string)
if cliAdv != "" {
// Validate ClientAdvertise syntax
if _, _, err := parseHostPort(cliAdv, 0); err != nil {
return nil, fmt.Errorf("invalid ClientAdvertise value of %s, err=%v", cliAdv, err)
}
}
diffOpts = append(diffOpts, &clientAdvertiseOption{newValue: cliAdv})
case "accounts":
diffOpts = append(diffOpts, &accountsOption{})
case "resolver", "accountresolver", "accountsresolver":
// We can't move from no resolver to one. So check for that.
if (oldValue == nil && newValue != nil) ||
(oldValue != nil && newValue == nil) {
return nil, fmt.Errorf("config reload does not support moving to or from an account resolver")
}
diffOpts = append(diffOpts, &accountsOption{})
case "accountresolvertlsconfig":
diffOpts = append(diffOpts, &accountsOption{})
case "gateway":
// Not supported for now, but report warning if configuration of gateway
// is actually changed so that user knows that it won't take effect.
// Any deep-equal is likely to fail for when there is a TLSConfig. so
// remove for the test.
tmpOld := oldValue.(GatewayOpts)
tmpNew := newValue.(GatewayOpts)
tmpOld.TLSConfig = nil
tmpNew.TLSConfig = nil
tmpOld.tlsConfigOpts = nil
tmpNew.tlsConfigOpts = nil
// Need to do the same for remote gateways' TLS configs.
// But we can't just set remotes' TLSConfig to nil otherwise this
// would lose the real TLS configuration.
tmpOld.Gateways = copyRemoteGWConfigsWithoutTLSConfig(tmpOld.Gateways)
tmpNew.Gateways = copyRemoteGWConfigsWithoutTLSConfig(tmpNew.Gateways)
// If there is really a change prevents reload.
if !reflect.DeepEqual(tmpOld, tmpNew) {
// See TODO(ik) note below about printing old/new values.
return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
field.Name, oldValue, newValue)
}
case "leafnode":
// Similar to gateways
tmpOld := oldValue.(LeafNodeOpts)
tmpNew := newValue.(LeafNodeOpts)
tmpOld.TLSConfig = nil
tmpNew.TLSConfig = nil
tmpOld.tlsConfigOpts = nil
tmpNew.tlsConfigOpts = nil
// Need to do the same for remote leafnodes' TLS configs.
// But we can't just set remotes' TLSConfig to nil otherwise this
// would lose the real TLS configuration.
tmpOld.Remotes = copyRemoteLNConfigForReloadCompare(tmpOld.Remotes)
tmpNew.Remotes = copyRemoteLNConfigForReloadCompare(tmpNew.Remotes)
// Special check for leafnode remotes changes which are not supported right now.
leafRemotesChanged := func(a, b LeafNodeOpts) bool {
if len(a.Remotes) != len(b.Remotes) {
return true
}
// Check whether all remotes URLs are still the same.
for _, oldRemote := range a.Remotes {
var found bool
if oldRemote.LocalAccount == _EMPTY_ {
oldRemote.LocalAccount = globalAccountName
}
for _, newRemote := range b.Remotes {
// Bind to global account in case not defined.
if newRemote.LocalAccount == _EMPTY_ {
newRemote.LocalAccount = globalAccountName
}
if reflect.DeepEqual(oldRemote, newRemote) {
found = true
break
}
}
if !found {
return true
}
}
return false
}
// First check whether remotes changed at all. If they did not,
// skip them in the complete equal check.
if !leafRemotesChanged(tmpOld, tmpNew) {
tmpOld.Remotes = nil
tmpNew.Remotes = nil
}
// Special check for auth users to detect changes.
// If anything is off will fall through and fail below.
// If we detect they are semantically the same we nil them out
// to pass the check below.
if tmpOld.Users != nil || tmpNew.Users != nil {
if len(tmpOld.Users) == len(tmpNew.Users) {
oua := make(map[string]*User, len(tmpOld.Users))
nua := make(map[string]*User, len(tmpOld.Users))
for _, u := range tmpOld.Users {
oua[u.Username] = u
}
for _, u := range tmpNew.Users {
nua[u.Username] = u
}
same := true
for uname, u := range oua {
// If we can not find new one with same name, drop through to fail.
nu, ok := nua[uname]
if !ok {
same = false
break
}
// If username or password or account different break.
if u.Username != nu.Username || u.Password != nu.Password || u.Account.GetName() != nu.Account.GetName() {
same = false
break
}
}
// We can nil out here.
if same {
tmpOld.Users, tmpNew.Users = nil, nil
}
}
}
// If there is really a change prevents reload.
if !reflect.DeepEqual(tmpOld, tmpNew) {
// See TODO(ik) note below about printing old/new values.
return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
field.Name, oldValue, newValue)
}
case "jetstream":
new := newValue.(bool)
old := oldValue.(bool)
if new != old {
diffOpts = append(diffOpts, &jetStreamOption{newValue: new})
}
// Mark whether JS will be disabled.
disableJS = !new
case "storedir":
new := newValue.(string)
old := oldValue.(string)
modified := new != old
// Check whether JS is being disabled and/or storage dir attempted to change.
if jsEnabled && modified {
if new == _EMPTY_ {
// This means that either JS is being disabled or it is using an temp dir.
// Allow the change but error in case JS was not disabled.
jsStoreDirChanged = true
} else {
return nil, fmt.Errorf("config reload not supported for jetstream storage directory")
}
}
case "jetstreammaxmemory", "jetstreammaxstore":
old := oldValue.(int64)
new := newValue.(int64)
// Check whether JS is being disabled and/or limits are being changed.
var (
modified = new != old
fromUnset = old == -1
fromSet = !fromUnset
toUnset = new == -1
toSet = !toUnset
)
if jsEnabled && modified {
// Cannot change limits from dynamic storage at runtime.
switch {
case fromSet && toUnset:
// Limits changed but it may mean that JS is being disabled,
// keep track of the change and error in case it is not.
switch optName {
case "jetstreammaxmemory":
jsMemLimitsChanged = true
case "jetstreammaxstore":
jsFileLimitsChanged = true
default:
return nil, fmt.Errorf("config reload not supported for jetstream max memory and store")
}
case fromUnset && toSet:
// Prevent changing from dynamic max memory / file at runtime.
return nil, fmt.Errorf("config reload not supported for jetstream dynamic max memory and store")
default:
return nil, fmt.Errorf("config reload not supported for jetstream max memory and store")
}
}
case "websocket":
// Similar to gateways
tmpOld := oldValue.(WebsocketOpts)
tmpNew := newValue.(WebsocketOpts)
tmpOld.TLSConfig = nil
tmpNew.TLSConfig = nil
// If there is really a change prevents reload.
if !reflect.DeepEqual(tmpOld, tmpNew) {
// See TODO(ik) note below about printing old/new values.
return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
field.Name, oldValue, newValue)
}
case "mqtt":
diffOpts = append(diffOpts, &mqttAckWaitReload{newValue: newValue.(MQTTOpts).AckWait})
diffOpts = append(diffOpts, &mqttMaxAckPendingReload{newValue: newValue.(MQTTOpts).MaxAckPending})
diffOpts = append(diffOpts, &mqttStreamReplicasReload{newValue: newValue.(MQTTOpts).StreamReplicas})
diffOpts = append(diffOpts, &mqttConsumerReplicasReload{newValue: newValue.(MQTTOpts).ConsumerReplicas})
diffOpts = append(diffOpts, &mqttConsumerMemoryStorageReload{newValue: newValue.(MQTTOpts).ConsumerMemoryStorage})
diffOpts = append(diffOpts, &mqttInactiveThresholdReload{newValue: newValue.(MQTTOpts).ConsumerInactiveThreshold})
// Nil out/set to 0 the options that we allow to be reloaded so that
// we only fail reload if some that we don't support are changed.
tmpOld := oldValue.(MQTTOpts)
tmpNew := newValue.(MQTTOpts)
tmpOld.TLSConfig, tmpOld.AckWait, tmpOld.MaxAckPending, tmpOld.StreamReplicas, tmpOld.ConsumerReplicas, tmpOld.ConsumerMemoryStorage = nil, 0, 0, 0, 0, false
tmpOld.ConsumerInactiveThreshold = 0
tmpNew.TLSConfig, tmpNew.AckWait, tmpNew.MaxAckPending, tmpNew.StreamReplicas, tmpNew.ConsumerReplicas, tmpNew.ConsumerMemoryStorage = nil, 0, 0, 0, 0, false
tmpNew.ConsumerInactiveThreshold = 0
if !reflect.DeepEqual(tmpOld, tmpNew) {
// See TODO(ik) note below about printing old/new values.
return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
field.Name, oldValue, newValue)
}
tmpNew.AckWait = newValue.(MQTTOpts).AckWait
tmpNew.MaxAckPending = newValue.(MQTTOpts).MaxAckPending
tmpNew.StreamReplicas = newValue.(MQTTOpts).StreamReplicas
tmpNew.ConsumerReplicas = newValue.(MQTTOpts).ConsumerReplicas
tmpNew.ConsumerMemoryStorage = newValue.(MQTTOpts).ConsumerMemoryStorage
tmpNew.ConsumerInactiveThreshold = newValue.(MQTTOpts).ConsumerInactiveThreshold
case "connecterrorreports":
diffOpts = append(diffOpts, &connectErrorReports{newValue: newValue.(int)})
case "reconnecterrorreports":
diffOpts = append(diffOpts, &reconnectErrorReports{newValue: newValue.(int)})
case "nolog", "nosigs":
// Ignore NoLog and NoSigs options since they are not parsed and only used in
// testing.
continue
case "disableshortfirstping":
newOpts.DisableShortFirstPing = oldValue.(bool)
continue
case "maxtracedmsglen":
diffOpts = append(diffOpts, &maxTracedMsgLenOption{newValue: newValue.(int)})
case "port":
// check to see if newValue == 0 and continue if so.
if newValue == 0 {
// ignore RANDOM_PORT
continue
}
fallthrough
case "noauthuser":
if oldValue != _EMPTY_ && newValue == _EMPTY_ {
for _, user := range newOpts.Users {
if user.Username == oldValue {
return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
field.Name, oldValue, newValue)
}
}
} else {
return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
field.Name, oldValue, newValue)
}
case "systemaccount":
if oldValue != DEFAULT_SYSTEM_ACCOUNT || newValue != _EMPTY_ {
return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
field.Name, oldValue, newValue)
}
case "ocspconfig":
diffOpts = append(diffOpts, &ocspOption{newValue: newValue.(*OCSPConfig)})
default:
// TODO(ik): Implement String() on those options to have a nice print.
// %v is difficult to figure what's what, %+v print private fields and
// would print passwords. Tried json.Marshal but it is too verbose for
// the URL array.
// Bail out if attempting to reload any unsupported options.
return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
field.Name, oldValue, newValue)
}
}
// If not disabling JS but limits have changed then it is an error.
if !disableJS {
if jsMemLimitsChanged || jsFileLimitsChanged {
return nil, fmt.Errorf("config reload not supported for jetstream max memory and max store")
}
if jsStoreDirChanged {
return nil, fmt.Errorf("config reload not supported for jetstream storage dir")
}
}
return diffOpts, nil
}
func copyRemoteGWConfigsWithoutTLSConfig(current []*RemoteGatewayOpts) []*RemoteGatewayOpts {
l := len(current)
if l == 0 {
return nil
}
rgws := make([]*RemoteGatewayOpts, 0, l)
for _, rcfg := range current {
cp := *rcfg
cp.TLSConfig = nil
cp.tlsConfigOpts = nil
rgws = append(rgws, &cp)
}
return rgws
}
func copyRemoteLNConfigForReloadCompare(current []*RemoteLeafOpts) []*RemoteLeafOpts {
l := len(current)
if l == 0 {
return nil
}
rlns := make([]*RemoteLeafOpts, 0, l)
for _, rcfg := range current {
cp := *rcfg
cp.TLSConfig = nil
cp.tlsConfigOpts = nil
// This is set only when processing a CONNECT, so reset here so that we
// don't fail the DeepEqual comparison.
cp.TLS = false
// For now, remove DenyImports/Exports since those get modified at runtime
// to add JS APIs.
cp.DenyImports, cp.DenyExports = nil, nil
rlns = append(rlns, &cp)
}
return rlns
}
func (s *Server) applyOptions(ctx *reloadContext, opts []option) {
var (
reloadLogging = false
reloadAuth = false
reloadClusterPerms = false
reloadClientTrcLvl = false
reloadJetstream = false
jsEnabled = false
reloadTLS = false
isStatszChange = false
)
for _, opt := range opts {
opt.Apply(s)
if opt.IsLoggingChange() {
reloadLogging = true
}
if opt.IsTraceLevelChange() {
reloadClientTrcLvl = true
}
if opt.IsAuthChange() {
reloadAuth = true
}
if opt.IsTLSChange() {
reloadTLS = true
}
if opt.IsClusterPermsChange() {
reloadClusterPerms = true
}
if opt.IsJetStreamChange() {
reloadJetstream = true
jsEnabled = opt.(*jetStreamOption).newValue
}
if opt.IsStatszChange() {
isStatszChange = true
}
}
if reloadLogging {
s.ConfigureLogger()
}
if reloadClientTrcLvl {
s.reloadClientTraceLevel()
}
if reloadAuth {
s.reloadAuthorization()
}
if reloadClusterPerms {
s.reloadClusterPermissions(ctx.oldClusterPerms)
}
if reloadJetstream {
if !jsEnabled {
s.DisableJetStream()
} else if !s.JetStreamEnabled() {
if err := s.restartJetStream(); err != nil {
s.Warnf("Can't start JetStream: %v", err)
}
}
// Make sure to reset the internal loop's version of JS.
s.resetInternalLoopInfo()
}
if isStatszChange {
s.sendStatszUpdate()
}
// For remote gateways and leafnodes, make sure that their TLS configuration
// is updated (since the config is "captured" early and changes would otherwise
// not be visible).
newOpts := s.getOpts()
if s.gateway.enabled {
s.gateway.updateRemotesTLSConfig(newOpts)
}
if len(newOpts.LeafNode.Remotes) > 0 {
s.updateRemoteLeafNodesTLSConfig(newOpts)
}
if reloadTLS {
// Restart OCSP monitoring.
if err := s.reloadOCSP(); err != nil {
s.Warnf("Can't restart OCSP Stapling: %v", err)
}
}
s.Noticef("Reloaded server configuration")
}
// This will send a reset to the internal send loop.
func (s *Server) resetInternalLoopInfo() {
var resetCh chan struct{}
s.mu.Lock()
if s.sys != nil {
// can't hold the lock as go routine reading it may be waiting for lock as well
resetCh = s.sys.resetCh
}
s.mu.Unlock()
if resetCh != nil {
resetCh <- struct{}{}
}
}
// Update all cached debug and trace settings for every client
func (s *Server) reloadClientTraceLevel() {
opts := s.getOpts()
if opts.NoLog {
return
}
// Create a list of all clients.
// Update their trace level when not holding server or gateway lock
s.mu.Lock()
clientCnt := 1 + len(s.clients) + len(s.grTmpClients) + len(s.routes) + len(s.leafs)
s.mu.Unlock()
s.gateway.RLock()
clientCnt += len(s.gateway.in) + len(s.gateway.outo)
s.gateway.RUnlock()
clients := make([]*client, 0, clientCnt)
s.mu.Lock()
if s.eventsEnabled() {
clients = append(clients, s.sys.client)
}
cMaps := []map[uint64]*client{s.clients, s.grTmpClients, s.routes, s.leafs}
for _, m := range cMaps {
for _, c := range m {
clients = append(clients, c)
}
}
s.mu.Unlock()
s.gateway.RLock()
for _, c := range s.gateway.in {
clients = append(clients, c)
}
clients = append(clients, s.gateway.outo...)
s.gateway.RUnlock()
for _, c := range clients {
// client.trace is commonly read while holding the lock
c.mu.Lock()
c.setTraceLevel()
c.mu.Unlock()
}
}
// reloadAuthorization reconfigures the server authorization settings,
// disconnects any clients who are no longer authorized, and removes any
// unauthorized subscriptions.
func (s *Server) reloadAuthorization() {
// This map will contain the names of accounts that have their streams
// import configuration changed.
awcsti := make(map[string]struct{})
checkJetStream := false
s.mu.Lock()
// This can not be changed for now so ok to check server's trustedKeys unlocked.
// If plain configured accounts, process here.
if s.trustedKeys == nil {
// We need to drain the old accounts here since we have something
// new configured. We do not want s.accounts to change since that would
// mean adding a lock to lookupAccount which is what we are trying to
// optimize for with the change from a map to a sync.Map.
oldAccounts := make(map[string]*Account)
s.accounts.Range(func(k, v interface{}) bool {
acc := v.(*Account)
acc.mu.Lock()
oldAccounts[acc.Name] = acc
// Need to clear out eventing timers since they close over this account and not the new one.
clearTimer(&acc.etmr)
clearTimer(&acc.ctmr)
acc.mu.Unlock()
s.accounts.Delete(k)
return true
})
s.gacc = nil
s.configureAccounts()
s.configureAuthorization()
s.mu.Unlock()
s.accounts.Range(func(k, v interface{}) bool {
newAcc := v.(*Account)
if acc, ok := oldAccounts[newAcc.Name]; ok {
// If account exist in latest config, "transfer" the account's
// sublist and client map to the new account.
acc.mu.RLock()
newAcc.mu.Lock()
if len(acc.clients) > 0 {
newAcc.clients = make(map[*client]struct{}, len(acc.clients))
for c := range acc.clients {
newAcc.clients[c] = struct{}{}
}
}
// Same for leafnodes
newAcc.lleafs = append([]*client(nil), acc.lleafs...)
newAcc.sl = acc.sl
if acc.rm != nil {
newAcc.rm = make(map[string]int32)
}
for k, v := range acc.rm {
newAcc.rm[k] = v
}
// Transfer internal client state. The configureAccounts call from above may have set up a new one.
// We need to use the old one, and the isid to not confuse internal subs.
newAcc.ic, newAcc.isid = acc.ic, acc.isid
// Transfer any JetStream state.
newAcc.js = acc.js
// Also transfer any internal accounting on different client types. We copy over all clients
// so need to copy this as well for proper accounting going forward.
newAcc.nrclients = acc.nrclients
newAcc.sysclients = acc.sysclients
newAcc.nleafs = acc.nleafs
newAcc.nrleafs = acc.nrleafs
// Process any reverse map entries.
if len(acc.imports.rrMap) > 0 {
newAcc.imports.rrMap = make(map[string][]*serviceRespEntry)
for k, v := range acc.imports.rrMap {
newAcc.imports.rrMap[k] = v
}
}
newAcc.mu.Unlock()
acc.mu.RUnlock()
// Check if current and new config of this account are same
// in term of stream imports.
if !acc.checkStreamImportsEqual(newAcc) {
awcsti[newAcc.Name] = struct{}{}
}
// We need to remove all old service import subs.
acc.removeAllServiceImportSubs()
newAcc.addAllServiceImportSubs()
}
return true
})
s.mu.Lock()
// Check if we had a default system account.
if s.sys != nil && s.sys.account != nil && !s.opts.NoSystemAccount {
s.accounts.Store(s.sys.account.Name, s.sys.account)
}
// Double check any JetStream configs.
checkJetStream = s.js != nil
} else if s.opts.AccountResolver != nil {
s.configureResolver()
if _, ok := s.accResolver.(*MemAccResolver); ok {
// Check preloads so we can issue warnings etc if needed.
s.checkResolvePreloads()
// With a memory resolver we want to do something similar to configured accounts.
// We will walk the accounts and delete them if they are no longer present via fetch.
// If they are present we will force a claim update to process changes.
s.accounts.Range(func(k, v interface{}) bool {
acc := v.(*Account)
// Skip global account.
if acc == s.gacc {
return true
}
accName := acc.GetName()
// Release server lock for following actions
s.mu.Unlock()
accClaims, claimJWT, _ := s.fetchAccountClaims(accName)
if accClaims != nil {
if err := s.updateAccountWithClaimJWT(acc, claimJWT); err != nil {
s.Noticef("Reloaded: deleting account [bad claims]: %q", accName)
s.accounts.Delete(k)
}
} else {
s.Noticef("Reloaded: deleting account [removed]: %q", accName)
s.accounts.Delete(k)
}
// Regrab server lock.
s.mu.Lock()
return true
})
}
}
var (
cclientsa [64]*client
cclients = cclientsa[:0]
clientsa [64]*client
clients = clientsa[:0]
routesa [64]*client
routes = routesa[:0]
)
// Gather clients that changed accounts. We will close them and they
// will reconnect, doing the right thing.
for _, client := range s.clients {
if s.clientHasMovedToDifferentAccount(client) {
cclients = append(cclients, client)
} else {
clients = append(clients, client)
}
}
for _, route := range s.routes {
routes = append(routes, route)
}
// Check here for any system/internal clients which will not be in the servers map of normal clients.
if s.sys != nil && s.sys.account != nil && !s.opts.NoSystemAccount {
s.accounts.Store(s.sys.account.Name, s.sys.account)
}
s.accounts.Range(func(k, v interface{}) bool {
acc := v.(*Account)
acc.mu.RLock()
// Check for sysclients accounting, ignore the system account.
if acc.sysclients > 0 && (s.sys == nil || s.sys.account != acc) {
for c := range acc.clients {
if c.kind != CLIENT && c.kind != LEAF {
clients = append(clients, c)
}
}
}
acc.mu.RUnlock()
return true
})
var resetCh chan struct{}
if s.sys != nil {
// can't hold the lock as go routine reading it may be waiting for lock as well
resetCh = s.sys.resetCh
}
s.mu.Unlock()
if resetCh != nil {
resetCh <- struct{}{}
}
// Check that publish retained messages sources are still allowed to publish.
s.mqttCheckPubRetainedPerms()
// Close clients that have moved accounts
for _, client := range cclients {
client.closeConnection(ClientClosed)
}
for _, c := range clients {
// Disconnect any unauthorized clients.
// Ignore internal clients.
if (c.kind == CLIENT || c.kind == LEAF) && !s.isClientAuthorized(c) {
c.authViolation()
continue
}
// Check to make sure account is correct.
c.swapAccountAfterReload()
// Remove any unauthorized subscriptions and check for account imports.
c.processSubsOnConfigReload(awcsti)
}
for _, route := range routes {
// Disconnect any unauthorized routes.
// Do this only for routes that were accepted, not initiated
// because in the later case, we don't have the user name/password
// of the remote server.
if !route.isSolicitedRoute() && !s.isRouterAuthorized(route) {
route.setNoReconnect()
route.authViolation()
}
}
if res := s.AccountResolver(); res != nil {
res.Reload()
}
// We will double check all JetStream configs on a reload.
if checkJetStream {
if err := s.enableJetStreamAccounts(); err != nil {
s.Errorf(err.Error())
}
}
}
// Returns true if given client current account has changed (or user
// no longer exist) in the new config, false if the user did not
// change accounts.
// Server lock is held on entry.
func (s *Server) clientHasMovedToDifferentAccount(c *client) bool {
var (
nu *NkeyUser
u *User
)
if c.opts.Nkey != "" {
if s.nkeys != nil {
nu = s.nkeys[c.opts.Nkey]
}
} else if c.opts.Username != "" {
if s.users != nil {
u = s.users[c.opts.Username]
}
} else {
return false
}
// Get the current account name
c.mu.Lock()
var curAccName string
if c.acc != nil {
curAccName = c.acc.Name
}
c.mu.Unlock()
if nu != nil && nu.Account != nil {
return curAccName != nu.Account.Name
} else if u != nil && u.Account != nil {
return curAccName != u.Account.Name
}
// user/nkey no longer exists.
return true
}
// reloadClusterPermissions reconfigures the cluster's permssions
// and set the permissions to all existing routes, sending an
// update INFO protocol so that remote can resend their local
// subs if needed, and sending local subs matching cluster's
// import subjects.
func (s *Server) reloadClusterPermissions(oldPerms *RoutePermissions) {
s.mu.Lock()
var (
infoJSON []byte
newPerms = s.opts.Cluster.Permissions
routes = make(map[uint64]*client, len(s.routes))
withNewProto int
)
// Get all connected routes
for i, route := range s.routes {
// Count the number of routes that can understand receiving INFO updates.
route.mu.Lock()
if route.opts.Protocol >= RouteProtoInfo {
withNewProto++
}
route.mu.Unlock()
routes[i] = route
}
// If new permissions is nil, then clear routeInfo import/export
if newPerms == nil {
s.routeInfo.Import = nil
s.routeInfo.Export = nil
} else {
s.routeInfo.Import = newPerms.Import
s.routeInfo.Export = newPerms.Export
}
// Regenerate route INFO
s.generateRouteInfoJSON()
infoJSON = s.routeInfoJSON
gacc := s.gacc
s.mu.Unlock()
// If there were no route, we are done
if len(routes) == 0 {
return
}
// If only older servers, simply close all routes and they will do the right
// thing on reconnect.
if withNewProto == 0 {
for _, route := range routes {
route.closeConnection(RouteRemoved)
}
return
}
// Fake clients to test cluster permissions
oldPermsTester := &client{}
oldPermsTester.setRoutePermissions(oldPerms)
newPermsTester := &client{}
newPermsTester.setRoutePermissions(newPerms)
var (
_localSubs [4096]*subscription
localSubs = _localSubs[:0]
subsNeedSUB []*subscription
subsNeedUNSUB []*subscription
deleteRoutedSubs []*subscription
)
// FIXME(dlc) - Change for accounts.
gacc.sl.localSubs(&localSubs, false)
// Go through all local subscriptions
for _, sub := range localSubs {
// Get all subs that can now be imported
subj := string(sub.subject)
couldImportThen := oldPermsTester.canImport(subj)
canImportNow := newPermsTester.canImport(subj)
if canImportNow {
// If we could not before, then will need to send a SUB protocol.
if !couldImportThen {
subsNeedSUB = append(subsNeedSUB, sub)
}
} else if couldImportThen {
// We were previously able to import this sub, but now
// we can't so we need to send an UNSUB protocol
subsNeedUNSUB = append(subsNeedUNSUB, sub)
}
}
for _, route := range routes {
route.mu.Lock()
// If route is to older server, simply close connection.
if route.opts.Protocol < RouteProtoInfo {
route.mu.Unlock()
route.closeConnection(RouteRemoved)
continue
}
route.setRoutePermissions(newPerms)
for _, sub := range route.subs {
// If we can't export, we need to drop the subscriptions that
// we have on behalf of this route.
subj := string(sub.subject)
if !route.canExport(subj) {
delete(route.subs, string(sub.sid))
deleteRoutedSubs = append(deleteRoutedSubs, sub)
}
}
// Send an update INFO, which will allow remote server to show
// our current route config in monitoring and resend subscriptions
// that we now possibly allow with a change of Export permissions.
route.enqueueProto(infoJSON)
// Now send SUB and UNSUB protocols as needed.
route.sendRouteSubProtos(subsNeedSUB, false, nil)
route.sendRouteUnSubProtos(subsNeedUNSUB, false, nil)
route.mu.Unlock()
}
// Remove as a batch all the subs that we have removed from each route.
// FIXME(dlc) - Change for accounts.
gacc.sl.RemoveBatch(deleteRoutedSubs)
}
// validateClusterOpts ensures the new ClusterOpts does not change host or
// port, which do not support reload.
func validateClusterOpts(old, new ClusterOpts) error {
if old.Host != new.Host {
return fmt.Errorf("config reload not supported for cluster host: old=%s, new=%s",
old.Host, new.Host)
}
if old.Port != new.Port {
return fmt.Errorf("config reload not supported for cluster port: old=%d, new=%d",
old.Port, new.Port)
}
// Validate Cluster.Advertise syntax
if new.Advertise != "" {
if _, _, err := parseHostPort(new.Advertise, 0); err != nil {
return fmt.Errorf("invalid Cluster.Advertise value of %s, err=%v", new.Advertise, err)
}
}
return nil
}
// diffRoutes diffs the old routes and the new routes and returns the ones that
// should be added and removed from the server.
func diffRoutes(old, new []*url.URL) (add, remove []*url.URL) {
// Find routes to remove.
removeLoop:
for _, oldRoute := range old {
for _, newRoute := range new {
if urlsAreEqual(oldRoute, newRoute) {
continue removeLoop
}
}
remove = append(remove, oldRoute)
}
// Find routes to add.
addLoop:
for _, newRoute := range new {
for _, oldRoute := range old {
if urlsAreEqual(oldRoute, newRoute) {
continue addLoop
}
}
add = append(add, newRoute)
}
return add, remove
}