Skip to content

Commit

Permalink
manager: Allow API listeners to be bound after starting the manager
Browse files Browse the repository at this point in the history 
This adds BindRemote and BindControl methods to Manager, which can be
used to specify the remote API and control API addresses after creating
or starting the manager. Raft has been updated to accept a new remote
address after being started. If The RemoteAPI and ControlAPI fields are
passed to manager.New, it is not necessary to call BindRemote or
BindControl explicitly.

Signed-off-by: Aaron Lehmann <[email protected]>
  • Loading branch information
@aaronlehmann
aaronlehmann committed Jan 7, 2017
1 parent d6766b0 commit 490ea4f
Show file tree
Hide file tree
Showing 5 changed files with 260 additions and 99 deletions.
242 changes: 152 additions & 90 deletions manager/manager.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ type Config struct {

// RemoteAPI is a listening address for serving the remote API, and
// an optional advertise address.
RemoteAPI RemoteAddrs
RemoteAPI *RemoteAddrs

// JoinRaft is an optional address of a node in an existing raft
// cluster to join.
Expand Down Expand Up @@ -111,8 +111,7 @@ type Config struct {
// This is the high-level object holding and initializing all the manager
// subsystems.
type Manager struct {
config *Config
listeners []net.Listener
config Config

caserver *ca.Server
dispatcher *dispatcher.Dispatcher
Expand All @@ -132,9 +131,15 @@ type Manager struct {

cancelFunc context.CancelFunc

mu sync.Mutex
mu sync.Mutex
addrMu sync.Mutex

started chan struct{}
stopped bool

remoteListener chan net.Listener
controlListener chan net.Listener
errServe chan error
}

type closeOnceListener struct {
Expand All @@ -152,21 +157,102 @@ func (l *closeOnceListener) Close() error {

// New creates a Manager which has not started to accept requests yet.
func New(config *Config) (*Manager, error) {
dispatcherConfig := dispatcher.DefaultConfig()
err := os.MkdirAll(config.StateDir, 0700)
if err != nil {
return nil, errors.Wrap(err, "failed to create state directory")
}

raftStateDir := filepath.Join(config.StateDir, "raft")
err = os.MkdirAll(raftStateDir, 0700)
if err != nil {
return nil, errors.Wrap(err, "failed to create raft state directory")
}

raftCfg := raft.DefaultNodeConfig()

if config.ElectionTick > 0 {
raftCfg.ElectionTick = int(config.ElectionTick)
}
if config.HeartbeatTick > 0 {
raftCfg.HeartbeatTick = int(config.HeartbeatTick)
}

dekRotator, err := NewRaftDEKManager(config.SecurityConfig.KeyWriter())
if err != nil {
return nil, err
}

newNodeOpts := raft.NodeOptions{
ID: config.SecurityConfig.ClientTLSCreds.NodeID(),
JoinAddr: config.JoinRaft,
Config: raftCfg,
StateDir: raftStateDir,
ForceNewCluster: config.ForceNewCluster,
TLSCredentials: config.SecurityConfig.ClientTLSCreds,
KeyRotator: dekRotator,
}
raftNode := raft.NewNode(newNodeOpts)

opts := []grpc.ServerOption{
grpc.Creds(config.SecurityConfig.ServerTLSCreds)}

m := &Manager{
config: *config,
caserver: ca.NewServer(raftNode.MemoryStore(), config.SecurityConfig),
dispatcher: dispatcher.New(raftNode, dispatcher.DefaultConfig()),
logbroker: logbroker.New(raftNode.MemoryStore()),
server: grpc.NewServer(opts...),
localserver: grpc.NewServer(opts...),
raftNode: raftNode,
started: make(chan struct{}),
dekRotator: dekRotator,
remoteListener: make(chan net.Listener, 1),
controlListener: make(chan net.Listener, 1),
errServe: make(chan error, 2),
}

if config.ControlAPI != "" {
m.config.ControlAPI = ""
if err := m.BindControl(config.ControlAPI); err != nil {
return nil, err
}
}

if config.RemoteAPI != nil {
m.config.RemoteAPI = nil
// The context isn't used in this case (before (*Manager).Run).
if err := m.BindRemote(context.Background(), *config.RemoteAPI); err != nil {
l := <-m.controlListener
l.Close()
return nil, err
}
}

return m, nil
}

// BindRemote binds a port for the remote API.
func (m *Manager) BindRemote(ctx context.Context, addrs RemoteAddrs) error {
m.addrMu.Lock()
defer m.addrMu.Unlock()

if m.config.RemoteAPI != nil {
return errors.New("manager already has remote API address")
}

// If an AdvertiseAddr was specified, we use that as our
// externally-reachable address.
advertiseAddr := config.RemoteAPI.AdvertiseAddr
advertiseAddr := addrs.AdvertiseAddr

var advertiseAddrPort string
if advertiseAddr == "" {
// Otherwise, we know we are joining an existing swarm. Use a
// wildcard address to trigger remote autodetection of our
// address.
var err error
_, advertiseAddrPort, err = net.SplitHostPort(config.RemoteAPI.ListenAddr)
_, advertiseAddrPort, err = net.SplitHostPort(addrs.ListenAddr)
if err != nil {
return nil, fmt.Errorf("missing or invalid listen address %s", config.RemoteAPI.ListenAddr)
return fmt.Errorf("missing or invalid listen address %s", addrs.ListenAddr)
}

// Even with an IPv6 listening address, it's okay to use
Expand All @@ -175,28 +261,42 @@ func New(config *Config) (*Manager, error) {
advertiseAddr = net.JoinHostPort("0.0.0.0", advertiseAddrPort)
}

err := os.MkdirAll(config.StateDir, 0700)
l, err := net.Listen("tcp", addrs.ListenAddr)
if err != nil {
return nil, errors.Wrap(err, "failed to create state directory")
return errors.Wrap(err, "failed to listen on remote API address")
}

raftStateDir := filepath.Join(config.StateDir, "raft")
err = os.MkdirAll(raftStateDir, 0700)
if err != nil {
return nil, errors.Wrap(err, "failed to create raft state directory")
if advertiseAddrPort == "0" {
advertiseAddr = l.Addr().String()
addrs.ListenAddr = advertiseAddr
}

var listeners []net.Listener
m.config.RemoteAPI = &addrs

m.raftNode.SetAddr(ctx, advertiseAddr)
m.remoteListener <- l

return nil
}

// BindControl binds a local socket for the control API.
func (m *Manager) BindControl(addr string) error {
m.addrMu.Lock()
defer m.addrMu.Unlock()

if m.config.ControlAPI != "" {
return errors.New("manager already has a control API address")
}
m.config.ControlAPI = addr

// don't create a socket directory if we're on windows. we used named pipe
if runtime.GOOS != "windows" {
err := os.MkdirAll(filepath.Dir(config.ControlAPI), 0700)
err := os.MkdirAll(filepath.Dir(addr), 0700)
if err != nil {
return nil, errors.Wrap(err, "failed to create socket directory")
return errors.Wrap(err, "failed to create socket directory")
}
}

l, err := xnet.ListenLocal(config.ControlAPI)
l, err := xnet.ListenLocal(addr)

// A unix socket may fail to bind if the file already
// exists. Try replacing the file.
Expand All @@ -209,69 +309,16 @@ func New(config *Config) (*Manager, error) {
unwrappedErr = sys.Err
}
if unwrappedErr == syscall.EADDRINUSE {
os.Remove(config.ControlAPI)
l, err = xnet.ListenLocal(config.ControlAPI)
os.Remove(addr)
l, err = xnet.ListenLocal(addr)
}
}
if err != nil {
return nil, errors.Wrap(err, "failed to listen on control API address")
}

listeners = append(listeners, l)

l, err = net.Listen("tcp", config.RemoteAPI.ListenAddr)
if err != nil {
return nil, errors.Wrap(err, "failed to listen on remote API address")
}
if advertiseAddrPort == "0" {
advertiseAddr = l.Addr().String()
config.RemoteAPI.ListenAddr = advertiseAddr
return errors.Wrap(err, "failed to listen on control API address")
}
listeners = append(listeners, l)

raftCfg := raft.DefaultNodeConfig()

if config.ElectionTick > 0 {
raftCfg.ElectionTick = int(config.ElectionTick)
}
if config.HeartbeatTick > 0 {
raftCfg.HeartbeatTick = int(config.HeartbeatTick)
}

dekRotator, err := NewRaftDEKManager(config.SecurityConfig.KeyWriter())
if err != nil {
return nil, err
}

newNodeOpts := raft.NodeOptions{
ID: config.SecurityConfig.ClientTLSCreds.NodeID(),
Addr: advertiseAddr,
JoinAddr: config.JoinRaft,
Config: raftCfg,
StateDir: raftStateDir,
ForceNewCluster: config.ForceNewCluster,
TLSCredentials: config.SecurityConfig.ClientTLSCreds,
KeyRotator: dekRotator,
}
raftNode := raft.NewNode(newNodeOpts)

opts := []grpc.ServerOption{
grpc.Creds(config.SecurityConfig.ServerTLSCreds)}

m := &Manager{
config: config,
listeners: listeners,
caserver: ca.NewServer(raftNode.MemoryStore(), config.SecurityConfig),
dispatcher: dispatcher.New(raftNode, dispatcherConfig),
logbroker: logbroker.New(raftNode.MemoryStore()),
server: grpc.NewServer(opts...),
localserver: grpc.NewServer(opts...),
raftNode: raftNode,
started: make(chan struct{}),
dekRotator: dekRotator,
}

return m, nil
m.controlListener <- l
return nil
}

// RemovedFromRaft returns a channel that's closed if the manager is removed
Expand All @@ -282,6 +329,12 @@ func (m *Manager) RemovedFromRaft() <-chan struct{} {

// Addr returns tcp address on which remote api listens.
func (m *Manager) Addr() string {
m.addrMu.Lock()
defer m.addrMu.Unlock()

if m.config.RemoteAPI == nil {
return ""
}
return m.config.RemoteAPI.ListenAddr
}

Expand Down Expand Up @@ -353,12 +406,17 @@ func (m *Manager) Run(parent context.Context) error {
// requests (it has no TLS information to put in the metadata map).
forwardAsOwnRequest := func(ctx context.Context) (context.Context, error) { return ctx, nil }
handleRequestLocally := func(ctx context.Context) (context.Context, error) {
var remoteAddr string
if m.config.RemoteAPI.AdvertiseAddr != "" {
remoteAddr = m.config.RemoteAPI.AdvertiseAddr
} else {
remoteAddr = m.config.RemoteAPI.ListenAddr
remoteAddr := "127.0.0.1:0"

m.addrMu.Lock()
if m.config.RemoteAPI != nil {
if m.config.RemoteAPI.AdvertiseAddr != "" {
remoteAddr = m.config.RemoteAPI.AdvertiseAddr
} else {
remoteAddr = m.config.RemoteAPI.ListenAddr
}
}
m.addrMu.Unlock()

creds := m.config.SecurityConfig.ClientTLSCreds

Expand Down Expand Up @@ -404,10 +462,8 @@ func (m *Manager) Run(parent context.Context) error {
healthServer.SetServingStatus("Raft", api.HealthCheckResponse_NOT_SERVING)
localHealthServer.SetServingStatus("ControlAPI", api.HealthCheckResponse_NOT_SERVING)

errServe := make(chan error, len(m.listeners))
for _, lis := range m.listeners {
go m.serveListener(ctx, errServe, lis)
}
go m.serveListener(ctx, m.remoteListener)
go m.serveListener(ctx, m.controlListener)

defer func() {
m.server.Stop()
Expand Down Expand Up @@ -455,7 +511,7 @@ func (m *Manager) Run(parent context.Context) error {
}

// wait for an error in serving.
err = <-errServe
err = <-m.errServe
m.mu.Lock()
if m.stopped {
m.mu.Unlock()
Expand Down Expand Up @@ -753,7 +809,13 @@ func (m *Manager) handleLeadershipEvents(ctx context.Context, leadershipCh chan
}

// serveListener serves a listener for local and non local connections.
func (m *Manager) serveListener(ctx context.Context, errServe chan error, l net.Listener) {
func (m *Manager) serveListener(ctx context.Context, lCh <-chan net.Listener) {
var l net.Listener
select {
case l = <-lCh:
case <-ctx.Done():
return
}
ctx = log.WithLogger(ctx, log.G(ctx).WithFields(
logrus.Fields{
"proto": l.Addr().Network(),
Expand All @@ -764,10 +826,10 @@ func (m *Manager) serveListener(ctx context.Context, errServe chan error, l net.
// we need to disallow double closes because UnixListener.Close
// can delete unix-socket file of newer listener. grpc calls
// Close twice indeed: in Serve and in Stop.
errServe <- m.localserver.Serve(&closeOnceListener{Listener: l})
m.errServe <- m.localserver.Serve(&closeOnceListener{Listener: l})
} else {
log.G(ctx).Info("Listening for connections")
errServe <- m.server.Serve(l)
m.errServe <- m.server.Serve(l)
}
}

Expand Down
4 changes: 2 additions & 2 deletions manager/manager_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ func TestManager(t *testing.T) {
assert.NoError(t, err)

m, err := New(&Config{
RemoteAPI: RemoteAddrs{ListenAddr: "127.0.0.1:0"},
RemoteAPI: &RemoteAddrs{ListenAddr: "127.0.0.1:0"},
ControlAPI: temp.Name(),
StateDir: stateDir,
SecurityConfig: managerSecurityConfig,
Expand Down Expand Up @@ -218,7 +218,7 @@ func TestManagerLockUnlock(t *testing.T) {
require.NoError(t, err)

m, err := New(&Config{
RemoteAPI: RemoteAddrs{ListenAddr: "127.0.0.1:0"},
RemoteAPI: &RemoteAddrs{ListenAddr: "127.0.0.1:0"},
ControlAPI: temp.Name(),
StateDir: stateDir,
SecurityConfig: managerSecurityConfig,
Expand Down
Loading

0 comments on commit 490ea4f

Please sign in to comment.