From d15f8cad3d588d1ea7f734fd4c86e41af1defe91 Mon Sep 17 00:00:00 2001 From: Aaron Lehmann Date: Wed, 21 Dec 2016 10:53:54 -0800 Subject: [PATCH] Convert code to use connectionbroker package By using connectionbroker instead of using remotes directly, subsystems like the agent can connect directly to the local manager when running on a node that acts as a manager. This avoids the need for the manager to expose a TCP port at all times. Signed-off-by: Aaron Lehmann --- agent/agent.go | 10 +++--- agent/agent_test.go | 5 +-- agent/config.go | 8 ++--- agent/resource.go | 4 +-- agent/session.go | 28 ++++++---------- ca/certificates.go | 72 ++++++++++++++++++++--------------------- ca/certificates_test.go | 36 ++++++++++----------- ca/config.go | 23 +++++++------ ca/config_test.go | 32 +++++++++--------- ca/testutils/cautils.go | 5 +-- manager/manager.go | 29 ++++++++++++++--- node/node.go | 64 +++++++++++++++++++++--------------- node/node_test.go | 4 +-- 13 files changed, 174 insertions(+), 146 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index 434bdd75f9..d8cc595474 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -336,12 +336,10 @@ func (a *Agent) handleSessionMessage(ctx context.Context, message *api.SessionMe seen := map[api.Peer]struct{}{} for _, manager := range message.Managers { if manager.Peer.Addr == "" { - log.G(ctx).WithField("manager.addr", manager.Peer.Addr). - Warnf("skipping bad manager address") continue } - a.config.Managers.Observe(*manager.Peer, int(manager.Weight)) + a.config.ConnBroker.Remotes().Observe(*manager.Peer, int(manager.Weight)) seen[*manager.Peer] = struct{}{} } @@ -358,9 +356,9 @@ func (a *Agent) handleSessionMessage(ctx context.Context, message *api.SessionMe } // prune managers not in list. - for peer := range a.config.Managers.Weights() { + for peer := range a.config.ConnBroker.Remotes().Weights() { if _, ok := seen[peer]; !ok { - a.config.Managers.Remove(peer) + a.config.ConnBroker.Remotes().Remove(peer) } } @@ -468,7 +466,7 @@ func (a *Agent) Publisher(ctx context.Context, subscriptionID string) (exec.LogP ) err = a.withSession(ctx, func(session *session) error { - publisher, err = api.NewLogBrokerClient(session.conn).PublishLogs(ctx) + publisher, err = api.NewLogBrokerClient(session.conn.ClientConn).PublishLogs(ctx) return err }) if err != nil { diff --git a/agent/agent_test.go b/agent/agent_test.go index 398a9887bd..55a641f799 100644 --- a/agent/agent_test.go +++ b/agent/agent_test.go @@ -8,6 +8,7 @@ import ( "github.com/docker/swarmkit/api" "github.com/docker/swarmkit/ca" "github.com/docker/swarmkit/ca/testutils" + "github.com/docker/swarmkit/connectionbroker" "github.com/docker/swarmkit/remotes" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -73,7 +74,7 @@ func TestAgentStartStop(t *testing.T) { agent, err := New(&Config{ Executor: &NoopExecutor{}, - Managers: remotes, + ConnBroker: connectionbroker.New(remotes), Credentials: agentSecurityConfig.ClientTLSCreds, DB: db, }) @@ -147,7 +148,7 @@ func agentTestEnv(t *testing.T) (*Agent, func()) { agent, err := New(&Config{ Executor: &NoopExecutor{}, - Managers: remotes, + ConnBroker: connectionbroker.New(remotes), Credentials: agentSecurityConfig.ClientTLSCreds, DB: db, }) diff --git a/agent/config.go b/agent/config.go index d62e15c8d5..de9359842e 100644 --- a/agent/config.go +++ b/agent/config.go @@ -4,7 +4,7 @@ import ( "github.com/boltdb/bolt" "github.com/docker/swarmkit/agent/exec" "github.com/docker/swarmkit/api" - "github.com/docker/swarmkit/remotes" + "github.com/docker/swarmkit/connectionbroker" "github.com/pkg/errors" "google.golang.org/grpc/credentials" ) @@ -14,9 +14,9 @@ type Config struct { // Hostname the name of host for agent instance. Hostname string - // Managers provides the manager backend used by the agent. It will be - // updated with managers weights as observed by the agent. - Managers remotes.Remotes + // ConnBroker provides a connection broker for retrieving gRPC + // connections to managers. + ConnBroker *connectionbroker.Broker // Executor specifies the executor to use for the agent. Executor exec.Executor diff --git a/agent/resource.go b/agent/resource.go index eca7564aa0..8e88d2cd65 100644 --- a/agent/resource.go +++ b/agent/resource.go @@ -30,7 +30,7 @@ type ResourceAllocator interface { func (r *resourceAllocator) AttachNetwork(ctx context.Context, id, target string, addresses []string) (string, error) { var taskID string if err := r.agent.withSession(ctx, func(session *session) error { - client := api.NewResourceAllocatorClient(session.conn) + client := api.NewResourceAllocatorClient(session.conn.ClientConn) r, err := client.AttachNetwork(ctx, &api.AttachNetworkRequest{ Config: &api.NetworkAttachmentConfig{ Target: target, @@ -53,7 +53,7 @@ func (r *resourceAllocator) AttachNetwork(ctx context.Context, id, target string // DetachNetwork deletes a network attachment. func (r *resourceAllocator) DetachNetwork(ctx context.Context, aID string) error { return r.agent.withSession(ctx, func(session *session) error { - client := api.NewResourceAllocatorClient(session.conn) + client := api.NewResourceAllocatorClient(session.conn.ClientConn) _, err := client.DetachNetwork(ctx, &api.DetachNetworkRequest{ AttachmentID: aID, }) diff --git a/agent/session.go b/agent/session.go index a8f657ffa3..fc1dca0db6 100644 --- a/agent/session.go +++ b/agent/session.go @@ -7,9 +7,9 @@ import ( "github.com/Sirupsen/logrus" "github.com/docker/swarmkit/api" + "github.com/docker/swarmkit/connectionbroker" "github.com/docker/swarmkit/log" "github.com/docker/swarmkit/protobuf/ptypes" - "github.com/docker/swarmkit/remotes" "golang.org/x/net/context" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -30,8 +30,7 @@ var ( // flow into the agent, such as task assignment, are called back into the // agent through errs, messages and tasks. type session struct { - conn *grpc.ClientConn - addr string + conn *connectionbroker.Conn agent *Agent sessionID string @@ -61,12 +60,7 @@ func newSession(ctx context.Context, agent *Agent, delay time.Duration, sessionI // TODO(stevvooe): Need to move connection management up a level or create // independent connection for log broker client. - peer, err := agent.config.Managers.Select() - if err != nil { - s.errs <- err - return s - } - cc, err := grpc.Dial(peer.Addr, + cc, err := agent.config.ConnBroker.Select( grpc.WithTransportCredentials(agent.config.Credentials), grpc.WithTimeout(dispatcherRPCTimeout), ) @@ -74,7 +68,6 @@ func newSession(ctx context.Context, agent *Agent, delay time.Duration, sessionI s.errs <- err return s } - s.addr = peer.Addr s.conn = cc go s.run(ctx, delay, description) @@ -127,7 +120,7 @@ func (s *session) start(ctx context.Context, description *api.NodeDescription) e // Need to run Session in a goroutine since there's no way to set a // timeout for an individual Recv call in a stream. go func() { - client := api.NewDispatcherClient(s.conn) + client := api.NewDispatcherClient(s.conn.ClientConn) stream, err = client.Session(sessionCtx, &api.SessionRequest{ Description: description, @@ -160,7 +153,7 @@ func (s *session) start(ctx context.Context, description *api.NodeDescription) e func (s *session) heartbeat(ctx context.Context) error { log.G(ctx).Debugf("(*session).heartbeat") - client := api.NewDispatcherClient(s.conn) + client := api.NewDispatcherClient(s.conn.ClientConn) heartbeat := time.NewTimer(1) // send out a heartbeat right away defer heartbeat.Stop() @@ -224,7 +217,7 @@ func (s *session) logSubscriptions(ctx context.Context) error { log := log.G(ctx).WithFields(logrus.Fields{"method": "(*session).logSubscriptions"}) log.Debugf("") - client := api.NewLogBrokerClient(s.conn) + client := api.NewLogBrokerClient(s.conn.ClientConn) subscriptions, err := client.ListenSubscriptions(ctx, &api.ListenSubscriptionsRequest{}) if err != nil { return err @@ -269,7 +262,7 @@ func (s *session) watch(ctx context.Context) error { err error ) - client := api.NewDispatcherClient(s.conn) + client := api.NewDispatcherClient(s.conn.ClientConn) for { // If this is the first time we're running the loop, or there was a reference mismatch // attempt to get the assignmentWatch @@ -344,7 +337,7 @@ func (s *session) watch(ctx context.Context) error { // sendTaskStatus uses the current session to send the status of a single task. func (s *session) sendTaskStatus(ctx context.Context, taskID string, status *api.TaskStatus) error { - client := api.NewDispatcherClient(s.conn) + client := api.NewDispatcherClient(s.conn.ClientConn) if _, err := client.UpdateTaskStatus(ctx, &api.UpdateTaskStatusRequest{ SessionID: s.sessionID, Updates: []*api.UpdateTaskStatusRequest_TaskStatusUpdate{ @@ -385,7 +378,7 @@ func (s *session) sendTaskStatuses(ctx context.Context, updates ...*api.UpdateTa return updates, ctx.Err() } - client := api.NewDispatcherClient(s.conn) + client := api.NewDispatcherClient(s.conn.ClientConn) n := batchSize if len(updates) < n { @@ -416,8 +409,7 @@ func (s *session) sendError(err error) { func (s *session) close() error { s.closeOnce.Do(func() { if s.conn != nil { - s.agent.config.Managers.ObserveIfExists(api.Peer{Addr: s.addr}, -remotes.DefaultObservationWeight) - s.conn.Close() + s.conn.Close(false) } close(s.closed) diff --git a/ca/certificates.go b/ca/certificates.go index 85b8bb9a6e..52537c89e4 100644 --- a/ca/certificates.go +++ b/ca/certificates.go @@ -22,8 +22,8 @@ import ( "github.com/cloudflare/cfssl/signer/local" "github.com/docker/go-events" "github.com/docker/swarmkit/api" + "github.com/docker/swarmkit/connectionbroker" "github.com/docker/swarmkit/ioutils" - "github.com/docker/swarmkit/remotes" "github.com/opencontainers/go-digest" "github.com/pkg/errors" "golang.org/x/net/context" @@ -169,6 +169,15 @@ func (rca *RootCA) RequestAndSaveNewCertificates(ctx context.Context, kw KeyWrit if err == nil { break } + + // If the first attempt fails, we should try a remote + // connection. The local node may be a manager that was + // demoted, so the local connection (which is preferred) may + // not work. If we are successful in renewing the certificate, + // the local connection will not be returned by the connection + // broker anymore. + config.ForceRemote = true + } if err != nil { return nil, err @@ -202,7 +211,7 @@ func (rca *RootCA) RequestAndSaveNewCertificates(ctx context.Context, kw KeyWrit var kekUpdate *KEKData for i := 0; i < 5; i++ { - kekUpdate, err = rca.getKEKUpdate(ctx, X509Cert, tlsKeyPair, config.Remotes) + kekUpdate, err = rca.getKEKUpdate(ctx, X509Cert, tlsKeyPair, config.ConnBroker) if err == nil { break } @@ -218,7 +227,7 @@ func (rca *RootCA) RequestAndSaveNewCertificates(ctx context.Context, kw KeyWrit return &tlsKeyPair, nil } -func (rca *RootCA) getKEKUpdate(ctx context.Context, cert *x509.Certificate, keypair tls.Certificate, r remotes.Remotes) (*KEKData, error) { +func (rca *RootCA) getKEKUpdate(ctx context.Context, cert *x509.Certificate, keypair tls.Certificate, connBroker *connectionbroker.Broker) (*KEKData, error) { var managerRole bool for _, ou := range cert.Subject.OrganizationalUnit { if ou == ManagerRole { @@ -229,25 +238,25 @@ func (rca *RootCA) getKEKUpdate(ctx context.Context, cert *x509.Certificate, key if managerRole { mtlsCreds := credentials.NewTLS(&tls.Config{ServerName: CARole, RootCAs: rca.Pool, Certificates: []tls.Certificate{keypair}}) - conn, peer, err := getGRPCConnection(mtlsCreds, r) + conn, err := getGRPCConnection(mtlsCreds, connBroker, false) if err != nil { return nil, err } - defer conn.Close() - client := api.NewCAClient(conn) + client := api.NewCAClient(conn.ClientConn) ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() response, err := client.GetUnlockKey(ctx, &api.GetUnlockKeyRequest{}) if err != nil { if grpc.Code(err) == codes.Unimplemented { // if the server does not support keks, return as if no encryption key was specified + conn.Close(true) return &KEKData{}, nil } - r.Observe(peer, -remotes.DefaultObservationWeight) + conn.Close(false) return nil, err } - r.Observe(peer, remotes.DefaultObservationWeight) + conn.Close(true) return &KEKData{KEK: response.UnlockKey, Version: response.Version.Index}, nil } @@ -440,45 +449,33 @@ func GetLocalRootCA(paths CertPaths) (RootCA, error) { return NewRootCA(cert, key, DefaultNodeCertExpiration) } -func getGRPCConnection(creds credentials.TransportCredentials, r remotes.Remotes) (*grpc.ClientConn, api.Peer, error) { - peer, err := r.Select() - if err != nil { - return nil, api.Peer{}, err - } - - opts := []grpc.DialOption{ +func getGRPCConnection(creds credentials.TransportCredentials, connBroker *connectionbroker.Broker, forceRemote bool) (*connectionbroker.Conn, error) { + dialOpts := []grpc.DialOption{ grpc.WithTransportCredentials(creds), grpc.WithTimeout(5 * time.Second), grpc.WithBackoffMaxDelay(5 * time.Second), } - - conn, err := grpc.Dial(peer.Addr, opts...) - if err != nil { - return nil, api.Peer{}, err + if forceRemote { + return connBroker.SelectRemote(dialOpts...) } - return conn, peer, nil + return connBroker.Select(dialOpts...) } // GetRemoteCA returns the remote endpoint's CA certificate -func GetRemoteCA(ctx context.Context, d digest.Digest, r remotes.Remotes) (RootCA, error) { +func GetRemoteCA(ctx context.Context, d digest.Digest, connBroker *connectionbroker.Broker) (RootCA, error) { // This TLS Config is intentionally using InsecureSkipVerify. We use the // digest instead to check the integrity of the CA certificate. insecureCreds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: true}) - conn, peer, err := getGRPCConnection(insecureCreds, r) + conn, err := getGRPCConnection(insecureCreds, connBroker, false) if err != nil { return RootCA{}, err } - defer conn.Close() - client := api.NewCAClient(conn) + client := api.NewCAClient(conn.ClientConn) ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() defer func() { - if err != nil { - r.Observe(peer, -remotes.DefaultObservationWeight) - return - } - r.Observe(peer, remotes.DefaultObservationWeight) + conn.Close(err == nil) }() response, err := client.GetRootCACertificate(ctx, &api.GetRootCACertificateRequest{}) if err != nil { @@ -558,20 +555,22 @@ func GetRemoteSignedCertificate(ctx context.Context, csr []byte, rootCAPool *x50 creds = credentials.NewTLS(&tls.Config{ServerName: CARole, RootCAs: rootCAPool}) } - conn, peer, err := getGRPCConnection(creds, config.Remotes) + conn, err := getGRPCConnection(creds, config.ConnBroker, config.ForceRemote) if err != nil { return nil, err } - defer conn.Close() // Create a CAClient to retrieve a new Certificate - caClient := api.NewNodeCAClient(conn) + caClient := api.NewNodeCAClient(conn.ClientConn) + + issueCtx, issueCancel := context.WithTimeout(ctx, 5*time.Second) + defer issueCancel() // Send the Request and retrieve the request token issueRequest := &api.IssueNodeCertificateRequest{CSR: csr, Token: config.Token, Availability: config.Availability} - issueResponse, err := caClient.IssueNodeCertificate(ctx, issueRequest) + issueResponse, err := caClient.IssueNodeCertificate(issueCtx, issueRequest) if err != nil { - config.Remotes.Observe(peer, -remotes.DefaultObservationWeight) + conn.Close(false) return nil, err } @@ -589,13 +588,14 @@ func GetRemoteSignedCertificate(ctx context.Context, csr []byte, rootCAPool *x50 defer cancel() statusResponse, err := caClient.NodeCertificateStatus(ctx, statusRequest) if err != nil { - config.Remotes.Observe(peer, -remotes.DefaultObservationWeight) + conn.Close(false) return nil, err } // If the certificate was issued, return if statusResponse.Status.State == api.IssuanceStateIssued { if statusResponse.Certificate == nil { + conn.Close(false) return nil, errors.New("no certificate in CertificateStatus response") } @@ -605,7 +605,7 @@ func GetRemoteSignedCertificate(ctx context.Context, csr []byte, rootCAPool *x50 // retry until the certificate gets updated per our // current request. if bytes.Equal(statusResponse.Certificate.CSR, csr) { - config.Remotes.Observe(peer, remotes.DefaultObservationWeight) + conn.Close(true) return statusResponse.Certificate.Certificate, nil } } diff --git a/ca/certificates_test.go b/ca/certificates_test.go index 0d234397fa..449e98ebb3 100644 --- a/ca/certificates_test.go +++ b/ca/certificates_test.go @@ -258,7 +258,7 @@ func TestGetRemoteCA(t *testing.T) { d, err := digest.Parse("sha256:" + mdStr) assert.NoError(t, err) - cert, err := ca.GetRemoteCA(tc.Context, d, tc.Remotes) + cert, err := ca.GetRemoteCA(tc.Context, d, tc.ConnBroker) assert.NoError(t, err) assert.NotNil(t, cert) } @@ -276,7 +276,7 @@ func TestGetRemoteCAInvalidHash(t *testing.T) { tc := testutils.NewTestCA(t) defer tc.Stop() - _, err := ca.GetRemoteCA(tc.Context, "sha256:2d2f968475269f0dde5299427cf74348ee1d6115b95c6e3f283e5a4de8da445b", tc.Remotes) + _, err := ca.GetRemoteCA(tc.Context, "sha256:2d2f968475269f0dde5299427cf74348ee1d6115b95c6e3f283e5a4de8da445b", tc.ConnBroker) assert.Error(t, err) } @@ -288,8 +288,8 @@ func TestRequestAndSaveNewCertificates(t *testing.T) { rca := ca.RootCA{Cert: tc.RootCA.Cert, Pool: tc.RootCA.Pool} cert, err := rca.RequestAndSaveNewCertificates(tc.Context, tc.KeyReadWriter, ca.CertificateRequestConfig{ - Token: tc.ManagerToken, - Remotes: tc.Remotes, + Token: tc.ManagerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, cert) @@ -306,8 +306,8 @@ func TestRequestAndSaveNewCertificates(t *testing.T) { // the worker token is also unencrypted cert, err = rca.RequestAndSaveNewCertificates(tc.Context, tc.KeyReadWriter, ca.CertificateRequestConfig{ - Token: tc.WorkerToken, - Remotes: tc.Remotes, + Token: tc.WorkerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, cert) @@ -330,8 +330,8 @@ func TestRequestAndSaveNewCertificates(t *testing.T) { _, err = rca.RequestAndSaveNewCertificates(tc.Context, tc.KeyReadWriter, ca.CertificateRequestConfig{ - Token: tc.ManagerToken, - Remotes: tc.Remotes, + Token: tc.ManagerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) @@ -345,8 +345,8 @@ func TestRequestAndSaveNewCertificates(t *testing.T) { // if it's a worker though, the key is always unencrypted, even though the manager key is encrypted _, err = rca.RequestAndSaveNewCertificates(tc.Context, tc.KeyReadWriter, ca.CertificateRequestConfig{ - Token: tc.WorkerToken, - Remotes: tc.Remotes, + Token: tc.WorkerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) _, _, err = unencryptedKeyReader.Read() @@ -412,8 +412,8 @@ func TestGetRemoteSignedCertificate(t *testing.T) { certs, err := ca.GetRemoteSignedCertificate(context.Background(), csr, tc.RootCA.Pool, ca.CertificateRequestConfig{ - Token: tc.ManagerToken, - Remotes: tc.Remotes, + Token: tc.ManagerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, certs) @@ -429,8 +429,8 @@ func TestGetRemoteSignedCertificate(t *testing.T) { // Test the expiration for an worker certificate certs, err = ca.GetRemoteSignedCertificate(tc.Context, csr, tc.RootCA.Pool, ca.CertificateRequestConfig{ - Token: tc.WorkerToken, - Remotes: tc.Remotes, + Token: tc.WorkerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, certs) @@ -452,8 +452,8 @@ func TestGetRemoteSignedCertificateNodeInfo(t *testing.T) { cert, err := ca.GetRemoteSignedCertificate(context.Background(), csr, tc.RootCA.Pool, ca.CertificateRequestConfig{ - Token: tc.WorkerToken, - Remotes: tc.Remotes, + Token: tc.WorkerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, cert) @@ -476,8 +476,8 @@ func TestGetRemoteSignedCertificateWithPending(t *testing.T) { go func() { _, err := ca.GetRemoteSignedCertificate(context.Background(), csr, tc.RootCA.Pool, ca.CertificateRequestConfig{ - Token: tc.WorkerToken, - Remotes: tc.Remotes, + Token: tc.WorkerToken, + ConnBroker: tc.ConnBroker, }) completed <- err }() diff --git a/ca/config.go b/ca/config.go index 1afa536946..d2664bd635 100644 --- a/ca/config.go +++ b/ca/config.go @@ -16,9 +16,9 @@ import ( "github.com/Sirupsen/logrus" cfconfig "github.com/cloudflare/cfssl/config" "github.com/docker/swarmkit/api" + "github.com/docker/swarmkit/connectionbroker" "github.com/docker/swarmkit/identity" "github.com/docker/swarmkit/log" - "github.com/docker/swarmkit/remotes" "github.com/opencontainers/go-digest" "github.com/pkg/errors" "google.golang.org/grpc/credentials" @@ -200,7 +200,7 @@ func getCAHashFromToken(token string) (digest.Digest, error) { } // DownloadRootCA tries to retrieve a remote root CA and matches the digest against the provided token. -func DownloadRootCA(ctx context.Context, paths CertPaths, token string, r remotes.Remotes) (RootCA, error) { +func DownloadRootCA(ctx context.Context, paths CertPaths, token string, connBroker *connectionbroker.Broker) (RootCA, error) { var rootCA RootCA // Get a digest for the optional CA hash string that we've been provided // If we were provided a non-empty string, and it is an invalid hash, return @@ -221,7 +221,7 @@ func DownloadRootCA(ctx context.Context, paths CertPaths, token string, r remote // just been demoted, for example). for i := 0; i != 5; i++ { - rootCA, err = GetRemoteCA(ctx, d, r) + rootCA, err = GetRemoteCA(ctx, d, connBroker) if err == nil { break } @@ -313,11 +313,16 @@ type CertificateRequestConfig struct { Token string // Availability allows a user to control the current scheduling status of a node Availability api.NodeSpec_Availability - // Remotes is the set of remote CAs. - Remotes remotes.Remotes + // ConnBroker provides connections to CAs. + ConnBroker *connectionbroker.Broker // Credentials provides transport credentials for communicating with the // remote server. Credentials credentials.TransportCredentials + // ForceRemote specifies that only a remote (TCP) connection should + // be used to request the certificate. This may be necessary in cases + // where the local node is running a manager, but is in the process of + // being demoted. + ForceRemote bool } // CreateSecurityConfig creates a new key and cert for this node, either locally @@ -380,7 +385,7 @@ func (rootCA RootCA) CreateSecurityConfig(ctx context.Context, krw *KeyReadWrite // RenewTLSConfigNow gets a new TLS cert and key, and updates the security config if provided. This is similar to // RenewTLSConfig, except while that monitors for expiry, and periodically renews, this renews once and is blocking -func RenewTLSConfigNow(ctx context.Context, s *SecurityConfig, r remotes.Remotes) error { +func RenewTLSConfigNow(ctx context.Context, s *SecurityConfig, connBroker *connectionbroker.Broker) error { s.renewalMu.Lock() defer s.renewalMu.Unlock() @@ -395,7 +400,7 @@ func RenewTLSConfigNow(ctx context.Context, s *SecurityConfig, r remotes.Remotes tlsKeyPair, err := rootCA.RequestAndSaveNewCertificates(ctx, s.KeyWriter(), CertificateRequestConfig{ - Remotes: r, + ConnBroker: connBroker, Credentials: s.ClientTLSCreds, }) if err != nil { @@ -437,7 +442,7 @@ func RenewTLSConfigNow(ctx context.Context, s *SecurityConfig, r remotes.Remotes // RenewTLSConfig will continuously monitor for the necessity of renewing the local certificates, either by // issuing them locally if key-material is available, or requesting them from a remote CA. -func RenewTLSConfig(ctx context.Context, s *SecurityConfig, remotes remotes.Remotes, renew <-chan struct{}) <-chan CertificateUpdate { +func RenewTLSConfig(ctx context.Context, s *SecurityConfig, connBroker *connectionbroker.Broker, renew <-chan struct{}) <-chan CertificateUpdate { updates := make(chan CertificateUpdate) go func() { @@ -501,7 +506,7 @@ func RenewTLSConfig(ctx context.Context, s *SecurityConfig, remotes remotes.Remo // ignore errors - it will just try again later var certUpdate CertificateUpdate - if err := RenewTLSConfigNow(ctx, s, remotes); err != nil { + if err := RenewTLSConfigNow(ctx, s, connBroker); err != nil { certUpdate.Err = err } else { certUpdate.Role = s.ClientTLSCreds.Role() diff --git a/ca/config_test.go b/ca/config_test.go index 6bcde60739..2e5bea9f46 100644 --- a/ca/config_test.go +++ b/ca/config_test.go @@ -25,7 +25,7 @@ func TestDownloadRootCASuccess(t *testing.T) { // Remove the CA cert os.RemoveAll(tc.Paths.RootCA.Cert) - rootCA, err := ca.DownloadRootCA(tc.Context, tc.Paths.RootCA, tc.WorkerToken, tc.Remotes) + rootCA, err := ca.DownloadRootCA(tc.Context, tc.Paths.RootCA, tc.WorkerToken, tc.ConnBroker) require.NoError(t, err) require.NotNil(t, rootCA.Pool) require.NotNil(t, rootCA.Cert) @@ -37,7 +37,7 @@ func TestDownloadRootCASuccess(t *testing.T) { os.RemoveAll(tc.Paths.RootCA.Cert) // downloading without a join token also succeeds - rootCA, err = ca.DownloadRootCA(tc.Context, tc.Paths.RootCA, "", tc.Remotes) + rootCA, err = ca.DownloadRootCA(tc.Context, tc.Paths.RootCA, "", tc.ConnBroker) require.NoError(t, err) require.NotNil(t, rootCA.Pool) require.NotNil(t, rootCA.Cert) @@ -54,7 +54,7 @@ func TestDownloadRootCAWrongCAHash(t *testing.T) { os.RemoveAll(tc.Paths.RootCA.Cert) // invalid token - _, err := ca.DownloadRootCA(tc.Context, tc.Paths.RootCA, "invalidtoken", tc.Remotes) + _, err := ca.DownloadRootCA(tc.Context, tc.Paths.RootCA, "invalidtoken", tc.ConnBroker) require.Error(t, err) require.Contains(t, err.Error(), "invalid join token") @@ -65,7 +65,7 @@ func TestDownloadRootCAWrongCAHash(t *testing.T) { os.RemoveAll(tc.Paths.RootCA.Cert) - _, err = ca.DownloadRootCA(tc.Context, tc.Paths.RootCA, replacementToken, tc.Remotes) + _, err = ca.DownloadRootCA(tc.Context, tc.Paths.RootCA, replacementToken, tc.ConnBroker) require.Error(t, err) require.Contains(t, err.Error(), "remote CA does not match fingerprint.") } @@ -79,8 +79,8 @@ func TestCreateSecurityConfigEmptyDir(t *testing.T) { krw := ca.NewKeyReadWriter(tc.Paths.Node, nil, nil) nodeConfig, err := tc.RootCA.CreateSecurityConfig(tc.Context, krw, ca.CertificateRequestConfig{ - Token: tc.WorkerToken, - Remotes: tc.Remotes, + Token: tc.WorkerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, nodeConfig) @@ -99,8 +99,8 @@ func TestCreateSecurityConfigNoCerts(t *testing.T) { krw := ca.NewKeyReadWriter(tc.Paths.Node, nil, nil) nodeConfig, err := tc.RootCA.CreateSecurityConfig(tc.Context, krw, ca.CertificateRequestConfig{ - Token: tc.WorkerToken, - Remotes: tc.Remotes, + Token: tc.WorkerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, nodeConfig) @@ -115,8 +115,8 @@ func TestCreateSecurityConfigNoCerts(t *testing.T) { assert.NoError(t, err) nodeConfig, err = rootCA.CreateSecurityConfig(tc.Context, krw, ca.CertificateRequestConfig{ - Token: tc.WorkerToken, - Remotes: tc.Remotes, + Token: tc.WorkerToken, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, nodeConfig) @@ -141,7 +141,7 @@ some random garbage\n nodeConfig, err := tc.RootCA.CreateSecurityConfig(tc.Context, krw, ca.CertificateRequestConfig{ - Remotes: tc.Remotes, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) @@ -167,7 +167,7 @@ some random garbage\n nodeConfig, err := tc.RootCA.CreateSecurityConfig(tc.Context, krw, ca.CertificateRequestConfig{ - Remotes: tc.Remotes, + ConnBroker: tc.ConnBroker, }) assert.NoError(t, err) assert.NotNil(t, nodeConfig) @@ -232,7 +232,7 @@ func TestRenewTLSConfigWorker(t *testing.T) { assert.NoError(t, err) renew := make(chan struct{}) - updates := ca.RenewTLSConfig(ctx, nodeConfig, tc.Remotes, renew) + updates := ca.RenewTLSConfig(ctx, nodeConfig, tc.ConnBroker, renew) select { case <-time.After(10 * time.Second): assert.Fail(t, "TestRenewTLSConfig timed-out") @@ -288,7 +288,7 @@ func TestRenewTLSConfigManager(t *testing.T) { // Get a new nodeConfig with a TLS cert that has 1 minute to live renew := make(chan struct{}) - updates := ca.RenewTLSConfig(ctx, nodeConfig, tc.Remotes, renew) + updates := ca.RenewTLSConfig(ctx, nodeConfig, tc.ConnBroker, renew) select { case <-time.After(10 * time.Second): assert.Fail(t, "TestRenewTLSConfig timed-out") @@ -350,7 +350,7 @@ func TestRenewTLSConfigWithNoNode(t *testing.T) { assert.NoError(t, err) renew := make(chan struct{}) - updates := ca.RenewTLSConfig(ctx, nodeConfig, tc.Remotes, renew) + updates := ca.RenewTLSConfig(ctx, nodeConfig, tc.ConnBroker, renew) select { case <-time.After(10 * time.Second): assert.Fail(t, "TestRenewTLSConfig timed-out") @@ -374,7 +374,7 @@ func TestForceRenewTLSConfig(t *testing.T) { assert.NoError(t, err) renew := make(chan struct{}, 1) - updates := ca.RenewTLSConfig(ctx, nodeConfig, tc.Remotes, renew) + updates := ca.RenewTLSConfig(ctx, nodeConfig, tc.ConnBroker, renew) renew <- struct{}{} select { case <-time.After(10 * time.Second): diff --git a/ca/testutils/cautils.go b/ca/testutils/cautils.go index 8474a11c0f..bb161b75d6 100644 --- a/ca/testutils/cautils.go +++ b/ca/testutils/cautils.go @@ -18,6 +18,7 @@ import ( "github.com/cloudflare/cfssl/signer/local" "github.com/docker/swarmkit/api" "github.com/docker/swarmkit/ca" + "github.com/docker/swarmkit/connectionbroker" "github.com/docker/swarmkit/identity" "github.com/docker/swarmkit/ioutils" "github.com/docker/swarmkit/manager/state/store" @@ -45,7 +46,7 @@ type TestCA struct { Conns []*grpc.ClientConn WorkerToken string ManagerToken string - Remotes remotes.Remotes + ConnBroker *connectionbroker.Broker KeyReadWriter *ca.KeyReadWriter } @@ -199,7 +200,7 @@ func NewTestCA(t *testing.T, krwGenerators ...func(ca.CertPaths) *ca.KeyReadWrit CAServer: caServer, WorkerToken: workerToken, ManagerToken: managerToken, - Remotes: remotes, + ConnBroker: connectionbroker.New(remotes), KeyReadWriter: krw, } } diff --git a/manager/manager.go b/manager/manager.go index f852d4f6a6..b1c65aa14b 100644 --- a/manager/manager.go +++ b/manager/manager.go @@ -1,6 +1,7 @@ package manager import ( + "crypto/tls" "crypto/x509" "encoding/pem" "fmt" @@ -16,6 +17,7 @@ import ( "github.com/docker/go-events" "github.com/docker/swarmkit/api" "github.com/docker/swarmkit/ca" + "github.com/docker/swarmkit/connectionbroker" "github.com/docker/swarmkit/log" "github.com/docker/swarmkit/manager/allocator" "github.com/docker/swarmkit/manager/controlapi" @@ -38,6 +40,7 @@ import ( "github.com/pkg/errors" "golang.org/x/net/context" "google.golang.org/grpc" + "google.golang.org/grpc/credentials" ) const ( @@ -557,9 +560,6 @@ func (m *Manager) updateKEK(ctx context.Context, cluster *api.Cluster) error { "node.role": ca.ManagerRole, }) - // we are our own peer from which we get certs - try to connect over the local socket - r := remotes.NewRemotes(api.Peer{Addr: m.Addr(), NodeID: nodeID}) - kekData := ca.KEKData{Version: cluster.Meta.Version.Index} for _, encryptionKey := range cluster.UnlockKeys { if encryptionKey.Subsystem == ca.ManagerRole { @@ -579,8 +579,27 @@ func (m *Manager) updateKEK(ctx context.Context, cluster *api.Cluster) error { // a best effort attempt to update the TLS certificate - if it fails, it'll be updated the next time it renews; // don't wait because it might take a bit go func() { - if err := ca.RenewTLSConfigNow(ctx, securityConfig, r); err != nil { - logger.WithError(err).Errorf("failed to download new TLS certificate after locking the cluster") + insecureCreds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: true}) + + conn, err := grpc.Dial( + m.config.ControlAPI, + grpc.WithTransportCredentials(insecureCreds), + grpc.WithDialer( + func(addr string, timeout time.Duration) (net.Conn, error) { + return xnet.DialTimeoutLocal(addr, timeout) + }), + ) + if err != nil { + logger.WithError(err).Error("failed to connect to local manager socket after locking the cluster") + return + } + + defer conn.Close() + + connBroker := connectionbroker.New(remotes.NewRemotes()) + connBroker.SetLocalConn(conn) + if err := ca.RenewTLSConfigNow(ctx, securityConfig, connBroker); err != nil { + logger.WithError(err).Error("failed to download new TLS certificate after locking the cluster") } }() } diff --git a/node/node.go b/node/node.go index cf1545c529..be776dc151 100644 --- a/node/node.go +++ b/node/node.go @@ -18,6 +18,7 @@ import ( "github.com/docker/swarmkit/agent/exec" "github.com/docker/swarmkit/api" "github.com/docker/swarmkit/ca" + "github.com/docker/swarmkit/connectionbroker" "github.com/docker/swarmkit/ioutils" "github.com/docker/swarmkit/log" "github.com/docker/swarmkit/manager" @@ -105,6 +106,7 @@ type Node struct { sync.RWMutex config *Config remotes *persistentRemotes + connBroker *connectionbroker.Broker role string roleCond *sync.Cond conn *grpc.ClientConn @@ -154,7 +156,6 @@ func New(c *Config) (*Node, error) { return nil, err } } - n := &Node{ remotes: newPersistentRemotes(stateFile, p...), role: ca.WorkerRole, @@ -174,6 +175,8 @@ func New(c *Config) (*Node, error) { } } + n.connBroker = connectionbroker.New(n.remotes) + n.roleCond = sync.NewCond(n.RLocker()) n.connCond = sync.NewCond(n.RLocker()) return n, nil @@ -261,7 +264,7 @@ func (n *Node) run(ctx context.Context) (err error) { } }() - updates := ca.RenewTLSConfig(ctx, securityConfig, n.remotes, forceCertRenewal) + updates := ca.RenewTLSConfig(ctx, securityConfig, n.connBroker, forceCertRenewal) go func() { for { select { @@ -368,17 +371,35 @@ func (n *Node) Err(ctx context.Context) error { } func (n *Node) runAgent(ctx context.Context, db *bolt.DB, creds credentials.TransportCredentials, ready chan<- struct{}) error { + waitCtx, waitCancel := context.WithCancel(ctx) + remotesCh := n.remotes.WaitSelect(ctx) + controlCh := n.ListenControlSocket(waitCtx) + +waitPeer: + for { + select { + case <-ctx.Done(): + break waitPeer + case <-remotesCh: + break waitPeer + case conn := <-controlCh: + if conn != nil { + break waitPeer + } + } + } + + waitCancel() + select { case <-ctx.Done(): - case <-n.remotes.WaitSelect(ctx): - } - if ctx.Err() != nil { return ctx.Err() + default: } a, err := agent.New(&agent.Config{ Hostname: n.config.Hostname, - Managers: n.remotes, + ConnBroker: n.connBroker, Executor: n.config.Executor, DB: db, NotifyNodeChange: n.notifyNodeChange, @@ -423,6 +444,7 @@ func (n *Node) setControlSocket(conn *grpc.ClientConn) { n.conn.Close() } n.conn = conn + n.connBroker.SetLocalConn(conn) n.connCond.Broadcast() n.Unlock() } @@ -447,15 +469,21 @@ func (n *Node) ListenControlSocket(ctx context.Context) <-chan *grpc.ClientConn defer close(done) defer n.RUnlock() for { - if ctx.Err() != nil { + select { + case <-ctx.Done(): return + default: } if conn == n.conn { n.connCond.Wait() continue } conn = n.conn - c <- conn + select { + case c <- conn: + case <-ctx.Done(): + return + } } }() return c @@ -532,7 +560,7 @@ func (n *Node) loadSecurityConfig(ctx context.Context) (*ca.SecurityConfig, erro } log.G(ctx).Debug("generated CA key and certificate") } else if err == ca.ErrNoLocalRootCA { // from previous error loading the root CA from disk - rootCA, err = ca.DownloadRootCA(ctx, paths.RootCA, n.config.JoinToken, n.remotes) + rootCA, err = ca.DownloadRootCA(ctx, paths.RootCA, n.config.JoinToken, n.connBroker) if err != nil { return nil, err } @@ -559,7 +587,7 @@ func (n *Node) loadSecurityConfig(ctx context.Context) (*ca.SecurityConfig, erro securityConfig, err = rootCA.CreateSecurityConfig(ctx, krw, ca.CertificateRequestConfig{ Token: n.config.JoinToken, Availability: n.config.Availability, - Remotes: n.remotes, + ConnBroker: n.connBroker, }) if err != nil { @@ -687,22 +715,6 @@ func (n *Node) runManager(ctx context.Context, securityConfig *ca.SecurityConfig go n.initManagerConnection(connCtx, ready) - // this happens only on initial start - if ready != nil { - go func(ready chan struct{}) { - select { - case <-ready: - addr, err := n.RemoteAPIAddr() - if err != nil { - log.G(ctx).WithError(err).Errorf("get remote api addr") - } else { - n.remotes.Observe(api.Peer{NodeID: n.NodeID(), Addr: addr}, remotes.DefaultObservationWeight) - } - case <-connCtx.Done(): - } - }(ready) - } - // wait for manager stop or for role change select { case <-done: diff --git a/node/node_test.go b/node/node_test.go index ef74912ef5..8ec421675b 100644 --- a/node/node_test.go +++ b/node/node_test.go @@ -101,7 +101,7 @@ func TestLoadSecurityConfigLoadFromDisk(t *testing.T) { tc := cautils.NewTestCA(t) defer tc.Stop() - peer, err := tc.Remotes.Select() + peer, err := tc.ConnBroker.Remotes().Select() require.NoError(t, err) // Load successfully with valid passphrase @@ -169,7 +169,7 @@ func TestLoadSecurityConfigDownloadAllCerts(t *testing.T) { tc := cautils.NewTestCA(t) defer tc.Stop() - peer, err := tc.Remotes.Select() + peer, err := tc.ConnBroker.Remotes().Select() require.NoError(t, err) node, err = New(&Config{