update swarmkit to b40ffde

Signed-off-by: jose-bigio <jose.bigio@docker.com>
2017-11-28 13:55:40 -08:00
parent 8916204beb
commit ceb3b42a7f
11 changed files with 77 additions and 11 deletions
@@ -108,7 +108,7 @@ github.com/stevvooe/continuity cd7a8e21e2b6f84799f5dd4b65faf49c8d3ee02d
 github.com/tonistiigi/fsutil 0ac4c11b053b9c5c7c47558f81f96c7100ce50fb

 # cluster
-github.com/docker/swarmkit dcd1f2d56afc08827d060fdb8ad222b00b1b6000
+github.com/docker/swarmkit b40ffde2b85d5165de3c809c6566b05234aa384a 
 github.com/gogo/protobuf v0.4
 github.com/cloudflare/cfssl 7fb22c8cba7ecaf98e4082d22d65800cf45e042a
 github.com/google/certificate-transparency d90e65c3a07988180c5b1ece71791c0b6506826e
@@ -4,7 +4,9 @@
 package connectionbroker

 import (
+	"net"
 	"sync"
+	"time"

 	"github.com/docker/swarmkit/api"
 	"github.com/docker/swarmkit/remotes"
@@ -60,9 +62,14 @@ func (b *Broker) SelectRemote(dialOpts ...grpc.DialOption) (*Conn, error) {
 		return nil, err
 	}

+	// gRPC dialer connects to proxy first. Provide a custom dialer here avoid that.
+	// TODO(anshul) Add an option to configure this.
 	dialOpts = append(dialOpts,
 		grpc.WithUnaryInterceptor(grpc_prometheus.UnaryClientInterceptor),
-		grpc.WithStreamInterceptor(grpc_prometheus.StreamClientInterceptor))
+		grpc.WithStreamInterceptor(grpc_prometheus.StreamClientInterceptor),
+		grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
+			return net.DialTimeout("tcp", addr, timeout)
+		}))

 	cc, err := grpc.Dial(peer.Addr, dialOpts...)
 	if err != nil {
@@ -58,6 +58,10 @@ func validateClusterSpec(spec *api.ClusterSpec) error {
 		}
 	}

+	if spec.Annotations.Name != store.DefaultClusterName {
+		return grpc.Errorf(codes.InvalidArgument, "modification of cluster name is not allowed")
+	}
+
 	return nil
 }

@@ -248,6 +248,29 @@ func (s *Server) UpdateNode(ctx context.Context, request *api.UpdateNodeRequest)
 	}, nil
 }

+func removeNodeAttachments(tx store.Tx, nodeID string) error {
+	// orphan the node's attached containers. if we don't do this, the
+	// network these attachments are connected to will never be removeable
+	tasks, err := store.FindTasks(tx, store.ByNodeID(nodeID))
+	if err != nil {
+		return err
+	}
+	for _, task := range tasks {
+		// if the task is an attachment, then we just delete it. the allocator
+		// will do the heavy lifting. basically, GetAttachment will return the
+		// attachment if that's the kind of runtime, or nil if it's not.
+		if task.Spec.GetAttachment() != nil {
+			// don't delete the task. instead, update it to `ORPHANED` so that
+			// the taskreaper will clean it up.
+			task.Status.State = api.TaskStateOrphaned
+			if err := store.UpdateTask(tx, task); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // RemoveNode removes a Node referenced by NodeID with the given NodeSpec.
 // - Returns NotFound if the Node is not found.
 // - Returns FailedPrecondition if the Node has manager role (and is part of the memberlist) or is not shut down.
@@ -276,7 +299,7 @@ func (s *Server) RemoveNode(ctx context.Context, request *api.RemoveNodeRequest)
 		}

 		// lookup the cluster
-		clusters, err := store.FindClusters(tx, store.ByName("default"))
+		clusters, err := store.FindClusters(tx, store.ByName(store.DefaultClusterName))
 		if err != nil {
 			return err
 		}
@@ -313,6 +336,10 @@ func (s *Server) RemoveNode(ctx context.Context, request *api.RemoveNodeRequest)
 			return err
 		}

+		if err := removeNodeAttachments(tx, request.NodeID); err != nil {
+			return err
+		}
+
 		return store.DeleteNode(tx, request.NodeID)
 	})
 	if err != nil {
@@ -404,7 +404,7 @@ func (m *Manager) Run(parent context.Context) error {
 		)

 		m.raftNode.MemoryStore().View(func(readTx store.ReadTx) {
-			clusters, err = store.FindClusters(readTx, store.ByName("default"))
+			clusters, err = store.FindClusters(readTx, store.ByName(store.DefaultClusterName))

 		})

@@ -73,7 +73,7 @@ func (g *Orchestrator) Run(ctx context.Context) error {
 	var err error
 	g.store.View(func(readTx store.ReadTx) {
 		var clusters []*api.Cluster
-		clusters, err = store.FindClusters(readTx, store.ByName("default"))
+		clusters, err = store.FindClusters(readTx, store.ByName(store.DefaultClusterName))

 		if len(clusters) != 1 {
 			return // just pick up the cluster when it is created.
@@ -17,7 +17,7 @@ import (
 // responds to changes in individual tasks (or nodes which run them).

 func (r *Orchestrator) initCluster(readTx store.ReadTx) error {
-	clusters, err := store.FindClusters(readTx, store.ByName("default"))
+	clusters, err := store.FindClusters(readTx, store.ByName(store.DefaultClusterName))
 	if err != nil {
 		return err
 	}
@@ -169,7 +169,10 @@ func (f *PluginFilter) Check(n *NodeInfo) bool {
 		}
 	}

-	if f.t.Spec.LogDriver != nil {
+	// It's possible that the LogDriver object does not carry a name, just some
+	// configuration options. In that case, the plugin filter shouldn't fail to
+	// schedule the task
+	if f.t.Spec.LogDriver != nil && f.t.Spec.LogDriver.Name != "none" && f.t.Spec.LogDriver.Name != "" {
 		// If there are no log driver types in the list at all, most likely this is
 		// an older daemon that did not report this information. In this case don't filter
 		if typeFound, exists := f.pluginExistsOnNode("Log", f.t.Spec.LogDriver.Name, nodePlugins); !exists && typeFound {
@@ -180,9 +180,12 @@ type NodeOptions struct {
 	ClockSource clock.Clock
 	// SendTimeout is the timeout on the sending messages to other raft
 	// nodes. Leave this as 0 to get the default value.
-	SendTimeout    time.Duration
-	TLSCredentials credentials.TransportCredentials
-	KeyRotator     EncryptionKeyRotator
+	SendTimeout time.Duration
+	// LargeSendTimeout is the timeout on the sending snapshots to other raft
+	// nodes. Leave this as 0 to get the default value.
+	LargeSendTimeout time.Duration
+	TLSCredentials   credentials.TransportCredentials
+	KeyRotator       EncryptionKeyRotator
 	// DisableStackDump prevents Run from dumping goroutine stacks when the
 	// store becomes stuck.
 	DisableStackDump bool
@@ -204,6 +207,11 @@ func NewNode(opts NodeOptions) *Node {
 	if opts.SendTimeout == 0 {
 		opts.SendTimeout = 2 * time.Second
 	}
+	if opts.LargeSendTimeout == 0 {
+		// a "slow" 100Mbps connection can send over 240MB data in 20 seconds
+		// which is well over the gRPC message limit of 128MB allowed by SwarmKit
+		opts.LargeSendTimeout = 20 * time.Second
+	}

 	raftStore := raft.NewMemoryStorage()

@@ -349,6 +357,7 @@ func (n *Node) initTransport() {
 	transportConfig := &transport.Config{
 		HeartbeatInterval: time.Duration(n.Config.ElectionTick) * n.opts.TickInterval,
 		SendTimeout:       n.opts.SendTimeout,
+		LargeSendTimeout:  n.opts.LargeSendTimeout,
 		Credentials:       n.opts.TLSCredentials,
 		Raft:              n,
 	}
@@ -133,7 +133,14 @@ func (p *peer) resolveAddr(ctx context.Context, id uint64) (string, error) {
 }

 func (p *peer) sendProcessMessage(ctx context.Context, m raftpb.Message) error {
-	ctx, cancel := context.WithTimeout(ctx, p.tr.config.SendTimeout)
+	timeout := p.tr.config.SendTimeout
+	// if a snapshot is being sent, set timeout to LargeSendTimeout because
+	// sending snapshots can take more time than other messages sent between peers.
+	// The same applies to AppendEntries as well, where messages can get large.
+	if m.Type == raftpb.MsgSnap || m.Type == raftpb.MsgApp {
+		timeout = p.tr.config.LargeSendTimeout
+	}
+	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 	_, err := api.NewRaftClient(p.conn()).ProcessRaftMessage(ctx, &api.ProcessRaftMessageRequest{Message: &m})
 	if grpc.Code(err) == codes.NotFound && grpc.ErrorDesc(err) == membership.ErrMemberRemoved.Error() {
@@ -3,6 +3,7 @@
 package transport

 import (
+	"net"
 	"sync"
 	"time"

@@ -35,6 +36,7 @@ type Raft interface {
 type Config struct {
 	HeartbeatInterval time.Duration
 	SendTimeout       time.Duration
+	LargeSendTimeout  time.Duration
 	Credentials       credentials.TransportCredentials
 	RaftID            string

@@ -347,6 +349,13 @@ func (t *Transport) dial(addr string) (*grpc.ClientConn, error) {
 		grpcOptions = append(grpcOptions, grpc.WithTimeout(t.config.SendTimeout))
 	}

+	// gRPC dialer connects to proxy first. Provide a custom dialer here avoid that.
+	// TODO(anshul) Add an option to configure this.
+	grpcOptions = append(grpcOptions,
+		grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
+			return net.DialTimeout("tcp", addr, timeout)
+		}))
+
 	cc, err := grpc.Dial(addr, grpcOptions...)
 	if err != nil {
 		return nil, err