kotovalexarian-likes-github/moby--moby
1
0
Fork 0
mirror of https://github.com/moby/moby.git synced 2022-11-09 12:21:53 -05:00
Code Releases Activity

Merge pull request #27698 from tonistiigi/cluster-config-refactor

Browse source 
Refactor swarm node config to struct
This commit is contained in:
Alexander Morozov 2016-10-25 13:29:55 -07:00 • committed by GitHub
commit 844a3166e5
1 changed files with 63 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

112
daemon/cluster/cluster.go
View file

@ -55,19 +55,6 @@ var ErrPendingSwarmExists = fmt.Errorf("This node is processing an existing join
// ErrSwarmJoinTimeoutReached is returned when cluster join could not complete before timeout was reached. // ErrSwarmJoinTimeoutReached is returned when cluster join could not complete before timeout was reached.
var ErrSwarmJoinTimeoutReached = fmt.Errorf("Timeout was reached before node was joined. The attempt to join the swarm will continue in the background. Use the \"docker info\" command to see the current swarm status of your node.") var ErrSwarmJoinTimeoutReached = fmt.Errorf("Timeout was reached before node was joined. The attempt to join the swarm will continue in the background. Use the \"docker info\" command to see the current swarm status of your node.")
type state struct {
// LocalAddr is this machine's local IP or hostname, if specified.
LocalAddr string
// RemoteAddr is the address that was given to "swarm join. It is used
// to find LocalAddr if necessary.
RemoteAddr string
// ListenAddr is the address we bind to, including a port.
ListenAddr string
// AdvertiseAddr is the address other nodes should connect to,
// including a port.
AdvertiseAddr string
}
// NetworkSubnetsProvider exposes functions for retrieving the subnets // NetworkSubnetsProvider exposes functions for retrieving the subnets
// of networks managed by Docker, so they can be filtered. // of networks managed by Docker, so they can be filtered.
type NetworkSubnetsProvider interface { type NetworkSubnetsProvider interface {
@ -99,11 +86,7 @@ type Cluster struct {
runtimeRoot string runtimeRoot string
config Config config Config
configEvent chan struct{} // todo: make this array and goroutine safe configEvent chan struct{} // todo: make this array and goroutine safe
localAddr string
actualLocalAddr string // after resolution, not persisted actualLocalAddr string // after resolution, not persisted
remoteAddr string
listenAddr string
advertiseAddr string
stop bool stop bool
err error err error
cancelDelay func() cancelDelay func()
@ -129,6 +112,26 @@ type node struct {
conn *grpc.ClientConn conn *grpc.ClientConn
client swarmapi.ControlClient client swarmapi.ControlClient
reconnectDelay time.Duration reconnectDelay time.Duration
config nodeStartConfig
}
// nodeStartConfig holds configuration needed to start a new node. Exported
// fields of this structure are saved to disk in json. Unexported fields
// contain data that shouldn't be persisted between daemon reloads.
type nodeStartConfig struct {
// LocalAddr is this machine's local IP or hostname, if specified.
LocalAddr string
// RemoteAddr is the address that was given to "swarm join". It is used
// to find LocalAddr if necessary.
RemoteAddr string
// ListenAddr is the address we bind to, including a port.
ListenAddr string
// AdvertiseAddr is the address other nodes should connect to,
// including a port.
AdvertiseAddr string
joinAddr string
forceNewCluster bool
joinToken string
} }
// New creates a new Cluster instance using provided config. // New creates a new Cluster instance using provided config.
@ -151,7 +154,7 @@ func New(config Config) (*Cluster, error) {
attachers: make(map[string]*attacher), attachers: make(map[string]*attacher),
} }
st, err := c.loadState() nodeConfig, err := c.loadState()
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
return c, nil return c, nil
@ -159,7 +162,7 @@ func New(config Config) (*Cluster, error) {
return nil, err return nil, err
} }
n, err := c.startNewNode(false, st.LocalAddr, st.RemoteAddr, st.ListenAddr, st.AdvertiseAddr, "", "") n, err := c.startNewNode(*nodeConfig)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -175,7 +178,7 @@ func New(config Config) (*Cluster, error) {
return c, nil return c, nil
} }
func (c *Cluster) loadState() (*state, error) { func (c *Cluster) loadState() (*nodeStartConfig, error) {
dt, err := ioutil.ReadFile(filepath.Join(c.root, stateFile)) dt, err := ioutil.ReadFile(filepath.Join(c.root, stateFile))
if err != nil { if err != nil {
return nil, err return nil, err
@ -187,20 +190,15 @@ func (c *Cluster) loadState() (*state, error) {
} }
return nil, err return nil, err
} }
var st state var st nodeStartConfig
if err := json.Unmarshal(dt, &st); err != nil { if err := json.Unmarshal(dt, &st); err != nil {
return nil, err return nil, err
} }
return &st, nil return &st, nil
} }
func (c *Cluster) saveState() error { func (c *Cluster) saveState(config nodeStartConfig) error {
dt, err := json.Marshal(state{ dt, err := json.Marshal(config)
LocalAddr: c.localAddr,
RemoteAddr: c.remoteAddr,
ListenAddr: c.listenAddr,
AdvertiseAddr: c.advertiseAddr,
})
if err != nil { if err != nil {
return err return err
} }
@ -233,7 +231,10 @@ func (c *Cluster) reconnectOnFailure(n *node) {
return return
} }
var err error var err error
n, err = c.startNewNode(false, c.localAddr, c.getRemoteAddress(), c.listenAddr, c.advertiseAddr, c.getRemoteAddress(), "") config := n.config
config.RemoteAddr = c.getRemoteAddress()
config.joinAddr = config.RemoteAddr
n, err = c.startNewNode(config)
if err != nil { if err != nil {
c.err = err c.err = err
close(n.done) close(n.done)
@ -242,17 +243,17 @@ func (c *Cluster) reconnectOnFailure(n *node) {
} }
} }
func (c *Cluster) startNewNode(forceNewCluster bool, localAddr, remoteAddr, listenAddr, advertiseAddr, joinAddr, joinToken string) (*node, error) { func (c *Cluster) startNewNode(conf nodeStartConfig) (*node, error) {
if err := c.config.Backend.IsSwarmCompatible(); err != nil { if err := c.config.Backend.IsSwarmCompatible(); err != nil {
return nil, err return nil, err
} }
actualLocalAddr := localAddr actualLocalAddr := conf.LocalAddr
if actualLocalAddr == "" { if actualLocalAddr == "" {
// If localAddr was not specified, resolve it automatically // If localAddr was not specified, resolve it automatically
// based on the route to joinAddr. localAddr can only be left // based on the route to joinAddr. localAddr can only be left
// empty on "join". // empty on "join".
listenHost, _, err := net.SplitHostPort(listenAddr) listenHost, _, err := net.SplitHostPort(conf.ListenAddr)
if err != nil { if err != nil {
return nil, fmt.Errorf("could not parse listen address: %v", err) return nil, fmt.Errorf("could not parse listen address: %v", err)
} }
@ -261,12 +262,12 @@ func (c *Cluster) startNewNode(forceNewCluster bool, localAddr, remoteAddr, list
if listenAddrIP == nil || !listenAddrIP.IsUnspecified() { if listenAddrIP == nil || !listenAddrIP.IsUnspecified() {
actualLocalAddr = listenHost actualLocalAddr = listenHost
} else { } else {
if remoteAddr == "" { if conf.RemoteAddr == "" {
// Should never happen except using swarms created by // Should never happen except using swarms created by
// old versions that didn't save remoteAddr. // old versions that didn't save remoteAddr.
remoteAddr = "8.8.8.8:53" conf.RemoteAddr = "8.8.8.8:53"
} }
conn, err := net.Dial("udp", remoteAddr) conn, err := net.Dial("udp", conf.RemoteAddr)
if err != nil { if err != nil {
return nil, fmt.Errorf("could not find local IP address: %v", err) return nil, fmt.Errorf("could not find local IP address: %v", err)
} }
@ -281,13 +282,13 @@ func (c *Cluster) startNewNode(forceNewCluster bool, localAddr, remoteAddr, list
c.stop = false c.stop = false
n, err := swarmnode.New(&swarmnode.Config{ n, err := swarmnode.New(&swarmnode.Config{
Hostname: c.config.Name, Hostname: c.config.Name,
ForceNewCluster: forceNewCluster, ForceNewCluster: conf.forceNewCluster,
ListenControlAPI: filepath.Join(c.runtimeRoot, controlSocket), ListenControlAPI: filepath.Join(c.runtimeRoot, controlSocket),
ListenRemoteAPI: listenAddr, ListenRemoteAPI: conf.ListenAddr,
AdvertiseRemoteAPI: advertiseAddr, AdvertiseRemoteAPI: conf.AdvertiseAddr,
JoinAddr: joinAddr, JoinAddr: conf.joinAddr,
StateDir: c.root, StateDir: c.root,
JoinToken: joinToken, JoinToken: conf.joinToken,
Executor: container.NewExecutor(c.config.Backend), Executor: container.NewExecutor(c.config.Backend),
HeartbeatTick: 1, HeartbeatTick: 1,
ElectionTick: 3, ElectionTick: 3,
@ -303,14 +304,11 @@ func (c *Cluster) startNewNode(forceNewCluster bool, localAddr, remoteAddr, list
Node: n, Node: n,
done: make(chan struct{}), done: make(chan struct{}),
reconnectDelay: initialReconnectDelay, reconnectDelay: initialReconnectDelay,
config: conf,
} }
c.node = node c.node = node
c.localAddr = localAddr
c.actualLocalAddr = actualLocalAddr // not saved c.actualLocalAddr = actualLocalAddr // not saved
c.remoteAddr = remoteAddr c.saveState(conf)
c.listenAddr = listenAddr
c.advertiseAddr = advertiseAddr
c.saveState()
c.config.Backend.SetClusterProvider(c) c.config.Backend.SetClusterProvider(c)
go func() { go func() {
@ -417,7 +415,12 @@ func (c *Cluster) Init(req types.InitRequest) (string, error) {
} }
// todo: check current state existing // todo: check current state existing
n, err := c.startNewNode(req.ForceNewCluster, localAddr, "", net.JoinHostPort(listenHost, listenPort), net.JoinHostPort(advertiseHost, advertisePort), "", "") n, err := c.startNewNode(nodeStartConfig{
forceNewCluster: req.ForceNewCluster,
LocalAddr: localAddr,
ListenAddr: net.JoinHostPort(listenHost, listenPort),
AdvertiseAddr: net.JoinHostPort(advertiseHost, advertisePort),
})
if err != nil { if err != nil {
c.Unlock() c.Unlock()
return "", err return "", err
@ -472,7 +475,13 @@ func (c *Cluster) Join(req types.JoinRequest) error {
} }
// todo: check current state existing // todo: check current state existing
n, err := c.startNewNode(false, "", req.RemoteAddrs[0], net.JoinHostPort(listenHost, listenPort), advertiseAddr, req.RemoteAddrs[0], req.JoinToken) n, err := c.startNewNode(nodeStartConfig{
RemoteAddr: req.RemoteAddrs[0],
ListenAddr: net.JoinHostPort(listenHost, listenPort),
AdvertiseAddr: advertiseAddr,
joinAddr: req.RemoteAddrs[0],
joinToken: req.JoinToken,
})
if err != nil { if err != nil {
c.Unlock() c.Unlock()
return err return err
@ -702,15 +711,18 @@ func (c *Cluster) GetLocalAddress() string {
func (c *Cluster) GetListenAddress() string { func (c *Cluster) GetListenAddress() string {
c.RLock() c.RLock()
defer c.RUnlock() defer c.RUnlock()
return c.listenAddr if c.node != nil {
return c.node.config.ListenAddr
}
return ""
} }
// GetAdvertiseAddress returns the remotely reachable address of this node. // GetAdvertiseAddress returns the remotely reachable address of this node.
func (c *Cluster) GetAdvertiseAddress() string { func (c *Cluster) GetAdvertiseAddress() string {
c.RLock() c.RLock()
defer c.RUnlock() defer c.RUnlock()
if c.advertiseAddr != "" { if c.node != nil && c.node.config.AdvertiseAddr != "" {
advertiseHost, _, _ := net.SplitHostPort(c.advertiseAddr) advertiseHost, _, _ := net.SplitHostPort(c.node.config.AdvertiseAddr)
return advertiseHost return advertiseHost
} }
return c.actualLocalAddr return c.actualLocalAddr