Merge pull request #25050 from aaronlehmann/vendor-swarmkit-rc5
Vendor swarmkit for 1.12.0-rc5
This commit is contained in:
commit
9fc2eee2b5
8 changed files with 126 additions and 45 deletions
|
@ -139,7 +139,7 @@ clone git github.com/docker/docker-credential-helpers v0.3.0
|
|||
clone git github.com/docker/containerd 0ac3cd1be170d180b2baed755e8f0da547ceb267
|
||||
|
||||
# cluster
|
||||
clone git github.com/docker/swarmkit 4d7e44321726f011d010cdb72d2230f5db2b604e
|
||||
clone git github.com/docker/swarmkit 9d4c2f73124e70f8fa85f9076635b827d17b109f
|
||||
clone git github.com/golang/mock bd3c8e81be01eef76d4b503f5e687d2d1354d2d9
|
||||
clone git github.com/gogo/protobuf 43a2e0b1c32252bfbbdf81f7faa7a88fb3fa4028
|
||||
clone git github.com/cloudflare/cfssl b895b0549c0ff676f92cf09ba971ae02bb41367b
|
||||
|
|
|
@ -1,11 +1,11 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"time"
|
||||
|
||||
"github.com/docker/swarmkit/agent/exec"
|
||||
"github.com/docker/swarmkit/api"
|
||||
"github.com/docker/swarmkit/api/equality"
|
||||
"github.com/docker/swarmkit/log"
|
||||
"golang.org/x/net/context"
|
||||
)
|
||||
|
@ -175,7 +175,7 @@ func (tm *taskManager) run(ctx context.Context) {
|
|||
case status := <-statusq:
|
||||
tm.task.Status = *status
|
||||
case task := <-tm.updateq:
|
||||
if tasksEqual(task, tm.task) {
|
||||
if equality.TasksEqualStable(task, tm.task) {
|
||||
continue // ignore the update
|
||||
}
|
||||
|
||||
|
@ -241,17 +241,3 @@ func (tm *taskManager) run(ctx context.Context) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// tasksEqual returns true if the tasks are functionaly equal, ignoring status,
|
||||
// version and other superfluous fields.
|
||||
//
|
||||
// This used to decide whether or not to propagate a task update to a controller.
|
||||
func tasksEqual(a, b *api.Task) bool {
|
||||
// shallow copy
|
||||
copyA, copyB := *a, *b
|
||||
|
||||
copyA.Status, copyB.Status = api.TaskStatus{}, api.TaskStatus{}
|
||||
copyA.Meta, copyB.Meta = api.Meta{}, api.Meta{}
|
||||
|
||||
return reflect.DeepEqual(©A, ©B)
|
||||
}
|
||||
|
|
21
vendor/src/github.com/docker/swarmkit/api/equality/equality.go
vendored
Normal file
21
vendor/src/github.com/docker/swarmkit/api/equality/equality.go
vendored
Normal file
|
@ -0,0 +1,21 @@
|
|||
package equality
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
|
||||
"github.com/docker/swarmkit/api"
|
||||
)
|
||||
|
||||
// TasksEqualStable returns true if the tasks are functionaly equal, ignoring status,
|
||||
// version and other superfluous fields.
|
||||
//
|
||||
// This used to decide whether or not to propagate a task update to a controller.
|
||||
func TasksEqualStable(a, b *api.Task) bool {
|
||||
// shallow copy
|
||||
copyA, copyB := *a, *b
|
||||
|
||||
copyA.Status, copyB.Status = api.TaskStatus{}, api.TaskStatus{}
|
||||
copyA.Meta, copyB.Meta = api.Meta{}, api.Meta{}
|
||||
|
||||
return reflect.DeepEqual(©A, ©B)
|
||||
}
|
|
@ -14,6 +14,7 @@ import (
|
|||
|
||||
"github.com/Sirupsen/logrus"
|
||||
"github.com/docker/swarmkit/api"
|
||||
"github.com/docker/swarmkit/api/equality"
|
||||
"github.com/docker/swarmkit/ca"
|
||||
"github.com/docker/swarmkit/log"
|
||||
"github.com/docker/swarmkit/manager/state"
|
||||
|
@ -29,7 +30,7 @@ const (
|
|||
DefaultHeartBeatPeriod = 5 * time.Second
|
||||
defaultHeartBeatEpsilon = 500 * time.Millisecond
|
||||
defaultGracePeriodMultiplier = 3
|
||||
defaultRateLimitPeriod = 16 * time.Second
|
||||
defaultRateLimitPeriod = 8 * time.Second
|
||||
|
||||
// maxBatchItems is the threshold of queued writes that should
|
||||
// trigger an actual transaction to commit them to the shared store.
|
||||
|
@ -572,20 +573,44 @@ func (d *Dispatcher) Tasks(r *api.TasksRequest, stream api.Dispatcher_TasksServe
|
|||
return err
|
||||
}
|
||||
|
||||
select {
|
||||
case event := <-nodeTasks:
|
||||
switch v := event.(type) {
|
||||
case state.EventCreateTask:
|
||||
tasksMap[v.Task.ID] = v.Task
|
||||
case state.EventUpdateTask:
|
||||
tasksMap[v.Task.ID] = v.Task
|
||||
case state.EventDeleteTask:
|
||||
delete(tasksMap, v.Task.ID)
|
||||
// bursty events should be processed in batches and sent out snapshot
|
||||
const modificationBatchLimit = 200
|
||||
const eventPausedGap = 50 * time.Millisecond
|
||||
var modificationCnt int
|
||||
// eventPaused is true when there have been modifications
|
||||
// but next event has not arrived within eventPausedGap
|
||||
eventPaused := false
|
||||
|
||||
for modificationCnt < modificationBatchLimit && !eventPaused {
|
||||
select {
|
||||
case event := <-nodeTasks:
|
||||
switch v := event.(type) {
|
||||
case state.EventCreateTask:
|
||||
tasksMap[v.Task.ID] = v.Task
|
||||
modificationCnt++
|
||||
case state.EventUpdateTask:
|
||||
if oldTask, exists := tasksMap[v.Task.ID]; exists {
|
||||
if equality.TasksEqualStable(oldTask, v.Task) {
|
||||
// this update should not trigger action at agent
|
||||
tasksMap[v.Task.ID] = v.Task
|
||||
continue
|
||||
}
|
||||
}
|
||||
tasksMap[v.Task.ID] = v.Task
|
||||
modificationCnt++
|
||||
case state.EventDeleteTask:
|
||||
delete(tasksMap, v.Task.ID)
|
||||
modificationCnt++
|
||||
}
|
||||
case <-time.After(eventPausedGap):
|
||||
if modificationCnt > 0 {
|
||||
eventPaused = true
|
||||
}
|
||||
case <-stream.Context().Done():
|
||||
return stream.Context().Err()
|
||||
case <-d.ctx.Done():
|
||||
return d.ctx.Err()
|
||||
}
|
||||
case <-stream.Context().Done():
|
||||
return stream.Context().Err()
|
||||
case <-d.ctx.Done():
|
||||
return d.ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -12,10 +12,13 @@ import (
|
|||
"github.com/docker/swarmkit/manager/dispatcher/heartbeat"
|
||||
)
|
||||
|
||||
const rateLimitCount = 3
|
||||
|
||||
type registeredNode struct {
|
||||
SessionID string
|
||||
Heartbeat *heartbeat.Heartbeat
|
||||
Registered time.Time
|
||||
Attempts int
|
||||
Node *api.Node
|
||||
Disconnect chan struct{} // signal to disconnect
|
||||
mu sync.Mutex
|
||||
|
@ -86,9 +89,14 @@ func (s *nodeStore) CheckRateLimit(id string) error {
|
|||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if existRn, ok := s.nodes[id]; ok {
|
||||
if time.Since(existRn.Registered) < s.rateLimitPeriod {
|
||||
return grpc.Errorf(codes.Unavailable, "node %s attempted registration too recently", id)
|
||||
if time.Since(existRn.Registered) > s.rateLimitPeriod {
|
||||
existRn.Attempts = 0
|
||||
}
|
||||
existRn.Attempts++
|
||||
if existRn.Attempts > rateLimitCount {
|
||||
return grpc.Errorf(codes.Unavailable, "node %s exceeded rate limit count of registrations", id)
|
||||
}
|
||||
existRn.Registered = time.Now()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
@ -97,14 +105,22 @@ func (s *nodeStore) CheckRateLimit(id string) error {
|
|||
func (s *nodeStore) Add(n *api.Node, expireFunc func()) *registeredNode {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var attempts int
|
||||
var registered time.Time
|
||||
if existRn, ok := s.nodes[n.ID]; ok {
|
||||
attempts = existRn.Attempts
|
||||
registered = existRn.Registered
|
||||
existRn.Heartbeat.Stop()
|
||||
delete(s.nodes, n.ID)
|
||||
}
|
||||
if registered.IsZero() {
|
||||
registered = time.Now()
|
||||
}
|
||||
rn := ®isteredNode{
|
||||
SessionID: identity.NewID(), // session ID is local to the dispatcher.
|
||||
Node: n,
|
||||
Registered: time.Now(),
|
||||
Registered: registered,
|
||||
Attempts: attempts,
|
||||
Disconnect: make(chan struct{}),
|
||||
}
|
||||
s.nodes[n.ID] = rn
|
||||
|
|
|
@ -3,7 +3,6 @@ package manager
|
|||
import (
|
||||
"crypto/x509"
|
||||
"encoding/pem"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
|
@ -121,16 +120,26 @@ func New(config *Config) (*Manager, error) {
|
|||
config.ProtoAddr["tcp"] = config.ProtoListener["tcp"].Addr().String()
|
||||
}
|
||||
|
||||
tcpAddr := config.ProtoAddr["tcp"]
|
||||
|
||||
if config.AdvertiseAddr != "" {
|
||||
tcpAddr = config.AdvertiseAddr
|
||||
}
|
||||
// If an AdvertiseAddr was specified, we use that as our
|
||||
// externally-reachable address.
|
||||
tcpAddr := config.AdvertiseAddr
|
||||
|
||||
if tcpAddr == "" {
|
||||
return nil, errors.New("no tcp listen address or listener provided")
|
||||
// Otherwise, we know we are joining an existing swarm. Use a
|
||||
// wildcard address to trigger remote autodetection of our
|
||||
// address.
|
||||
_, tcpAddrPort, err := net.SplitHostPort(config.ProtoAddr["tcp"])
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("missing or invalid listen address %s", config.ProtoAddr["tcp"])
|
||||
}
|
||||
|
||||
// Even with an IPv6 listening address, it's okay to use
|
||||
// 0.0.0.0 here. Any "unspecified" (wildcard) IP will
|
||||
// be substituted with the actual source address.
|
||||
tcpAddr = net.JoinHostPort("0.0.0.0", tcpAddrPort)
|
||||
}
|
||||
|
||||
// FIXME(aaronl): Remove this. It appears to be unused.
|
||||
dispatcherConfig.Addr = tcpAddr
|
||||
|
||||
err := os.MkdirAll(filepath.Dir(config.ProtoAddr["unix"]), 0700)
|
||||
|
|
|
@ -7,7 +7,12 @@ var (
|
|||
// Always check for readiness first.
|
||||
&ReadyFilter{},
|
||||
&ResourceFilter{},
|
||||
&PluginFilter{},
|
||||
|
||||
// TODO(stevvooe): Do not filter based on plugins since they are lazy
|
||||
// loaded in the engine. We can add this back when we can schedule
|
||||
// plugins in the future.
|
||||
// &PluginFilter{},
|
||||
|
||||
&ConstraintFilter{},
|
||||
}
|
||||
)
|
||||
|
|
|
@ -401,7 +401,9 @@ func (n *Node) Run(ctx context.Context) error {
|
|||
// restoring from the state, campaign to be the
|
||||
// leader.
|
||||
if !n.restored {
|
||||
if len(n.cluster.Members()) <= 1 {
|
||||
// Node ID should be in the progress list to Campaign
|
||||
_, ok := n.Node.Status().Progress[n.Config.ID]
|
||||
if len(n.cluster.Members()) <= 1 && ok {
|
||||
if err := n.Campaign(n.Ctx); err != nil {
|
||||
panic("raft: cannot campaign to be the leader on node restore")
|
||||
}
|
||||
|
@ -779,10 +781,27 @@ func (n *Node) LeaderAddr() (string, error) {
|
|||
|
||||
// registerNode registers a new node on the cluster memberlist
|
||||
func (n *Node) registerNode(node *api.RaftMember) error {
|
||||
if n.cluster.IsIDRemoved(node.RaftID) {
|
||||
return nil
|
||||
}
|
||||
|
||||
member := &membership.Member{}
|
||||
|
||||
if n.cluster.GetMember(node.RaftID) != nil || n.cluster.IsIDRemoved(node.RaftID) {
|
||||
// member already exists
|
||||
existingMember := n.cluster.GetMember(node.RaftID)
|
||||
if existingMember != nil {
|
||||
// Member already exists
|
||||
|
||||
// If the address is different from what we thought it was,
|
||||
// update it. This can happen if we just joined a cluster
|
||||
// and are adding ourself now with the remotely-reachable
|
||||
// address.
|
||||
if existingMember.Addr != node.Addr {
|
||||
member.RaftMember = node
|
||||
member.RaftClient = existingMember.RaftClient
|
||||
member.Conn = existingMember.Conn
|
||||
n.cluster.AddMember(member)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in a new issue