@@ -41,7 +41,7 @@ github.com/vishvananda/netlink 1e86b2bee5b6a7d377e4c02bb7f98209d6a7297c

 github.com/BurntSushi/toml f706d00e3de6abe700c994cdd545a1a4915af060

 github.com/samuel/go-zookeeper d0e0d8e11f318e000a8cc434616d69e329edc374

 github.com/deckarep/golang-set ef32fa3046d9f249d399f98ebaf9be944430fd1d

-github.com/coreos/etcd 824277cb3a577a0e8c829ca9ec557b973fe06d20

+github.com/coreos/etcd ea5389a79f40206170582c1ea076191b8622cb8e https://github.com/aaronlehmann/etcd # for https://github.com/coreos/etcd/pull/7830

 github.com/ugorji/go f1f1a805ed361a0e078bb537e4ea78cd37dcf065

 github.com/hashicorp/consul v0.5.2

 github.com/boltdb/bolt fff57c100f4dea1905678da7e90d92429dff2904

@@ -108,7 +108,7 @@ github.com/docker/containerd 9048e5e50717ea4497b757314bad98ea3763c145

 github.com/tonistiigi/fifo 1405643975692217d6720f8b54aeee1bf2cd5cf4

 # cluster

-github.com/docker/swarmkit 61a92e8ec074df5769decda985df4a3ab43c77eb

+github.com/docker/swarmkit 8f053c2030ebfc90f19f241fb7880e95b9761b7a

 github.com/gogo/protobuf 8d70fb3182befc465c4a1eac8ad4d38ff49778e2

 github.com/cloudflare/cfssl 7fb22c8cba7ecaf98e4082d22d65800cf45e042a

 github.com/google/certificate-transparency d90e65c3a07988180c5b1ece71791c0b6506826e

@@ -1154,6 +1154,10 @@ func (r *raft) addNode(id uint64) {

 	}

 	r.setProgress(id, 0, r.raftLog.lastIndex()+1)

+	// When a node is first added, we should mark it as recently active.

+	// Otherwise, CheckQuorum may cause us to step down if it is invoked

+	// before the added node has a chance to communicate with us.

+	r.prs[id].RecentActive = true

 }

 func (r *raft) removeNode(id uint64) {

@@ -1,6 +1,6 @@

 # [SwarmKit](https://github.com/docker/swarmkit)

-[![GoDoc](https://godoc.org/github.com/docker/swarmkit?status.png)](https://godoc.org/github.com/docker/swarmkit)

+[![GoDoc](https://godoc.org/github.com/docker/swarmkit?status.svg)](https://godoc.org/github.com/docker/swarmkit)

 [![Circle CI](https://circleci.com/gh/docker/swarmkit.svg?style=shield&circle-token=a7bf494e28963703a59de71cf19b73ad546058a7)](https://circleci.com/gh/docker/swarmkit)

 [![codecov.io](https://codecov.io/github/docker/swarmkit/coverage.svg?branch=master&token=LqD1dzTjsN)](https://codecov.io/github/docker/swarmkit?branch=master)

 [![Badge Badge](http://doyouevenbadge.com/github.com/docker/swarmkit)](http://doyouevenbadge.com/report/github.com/docker/swarmkit)

@@ -83,7 +83,7 @@ Requirements:

 -   Go 1.6 or higher

 -   A [working golang](https://golang.org/doc/code.html) environment

--   [Protobuf 3.x or higher] (https://developers.google.com/protocol-buffers/docs/downloads) to regenerate protocol buffer files (e.g. using `make generate`)

+-   [Protobuf 3.x or higher](https://developers.google.com/protocol-buffers/docs/downloads) to regenerate protocol buffer files (e.g. using `make generate`)

 *SwarmKit* is built in Go and leverages a standard project structure to work well with Go tooling.

 If you are new to Go, please see [BUILDING.md](BUILDING.md) for a more detailed guide.

@@ -426,14 +426,19 @@ func (w *worker) Listen(ctx context.Context, reporter StatusReporter) {

 }

 func (w *worker) startTask(ctx context.Context, tx *bolt.Tx, task *api.Task) error {

-	w.taskevents.Publish(task.Copy())

 	_, err := w.taskManager(ctx, tx, task) // side-effect taskManager creation.

 	if err != nil {

 		log.G(ctx).WithError(err).Error("failed to start taskManager")

+		// we ignore this error: it gets reported in the taskStatus within

+		// `newTaskManager`. We log it here and move on. If their is an

+		// attempted restart, the lack of taskManager will have this retry

+		// again.

+		return nil

 	}

-	// TODO(stevvooe): Add start method for taskmanager

+	// only publish if controller resolution was successful.

+	w.taskevents.Publish(task.Copy())

 	return nil

 }

@@ -464,7 +469,7 @@ func (w *worker) newTaskManager(ctx context.Context, tx *bolt.Tx, task *api.Task

 	}

 	if err != nil {

-		log.G(ctx).Error("controller resolution failed")

+		log.G(ctx).WithError(err).Error("controller resolution failed")

 		return nil, err

 	}

@@ -568,9 +573,14 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe

 		case v := <-ch:

 			task := v.(*api.Task)

 			if match(task) {

-				w.mu.Lock()

-				go w.taskManagers[task.ID].Logs(ctx, *subscription.Options, publisher)

-				w.mu.Unlock()

+				w.mu.RLock()

+				tm, ok := w.taskManagers[task.ID]

+				w.mu.RUnlock()

+				if !ok {

+					continue

+				}

+

+				go tm.Logs(ctx, *subscription.Options, publisher)

 			}

 		case <-ctx.Done():

 			return ctx.Err()

@@ -14,7 +14,6 @@ import (

 	"github.com/Sirupsen/logrus"

 	cfconfig "github.com/cloudflare/cfssl/config"

-	events "github.com/docker/go-events"

 	"github.com/docker/swarmkit/api"

 	"github.com/docker/swarmkit/connectionbroker"

 	"github.com/docker/swarmkit/identity"

@@ -51,13 +50,6 @@ const (

 	base36DigestLen = 50

 )

-// RenewTLSExponentialBackoff sets the exponential backoff when trying to renew TLS certificates that have expired

-var RenewTLSExponentialBackoff = events.ExponentialBackoffConfig{

-	Base:   time.Second * 5,

-	Factor: time.Second * 5,

-	Max:    1 * time.Hour,

-}

-

 // SecurityConfig is used to represent a node's security configuration. It includes information about

 // the RootCA and ServerTLSCreds/ClientTLSCreds transport authenticators to be used for MTLS

 type SecurityConfig struct {

@@ -468,96 +460,6 @@ func RenewTLSConfigNow(ctx context.Context, s *SecurityConfig, connBroker *conne

 	return s.updateTLSCredentials(tlsKeyPair, issuerInfo)

 }

-// 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, connBroker *connectionbroker.Broker, renew <-chan struct{}) <-chan CertificateUpdate {

-	updates := make(chan CertificateUpdate)

-

-	go func() {

-		var (

-			retry      time.Duration

-			forceRetry bool

-		)

-		expBackoff := events.NewExponentialBackoff(RenewTLSExponentialBackoff)

-		defer close(updates)

-		for {

-			ctx = log.WithModule(ctx, "tls")

-			log := log.G(ctx).WithFields(logrus.Fields{

-				"node.id":   s.ClientTLSCreds.NodeID(),

-				"node.role": s.ClientTLSCreds.Role(),

-			})

-			// Our starting default will be 5 minutes

-			retry = 5 * time.Minute

-

-			// Since the expiration of the certificate is managed remotely we should update our

-			// retry timer on every iteration of this loop.

-			// Retrieve the current certificate expiration information.

-			validFrom, validUntil, err := readCertValidity(s.KeyReader())

-			if err != nil {

-				// We failed to read the expiration, let's stick with the starting default

-				log.Errorf("failed to read the expiration of the TLS certificate in: %s", s.KeyReader().Target())

-

-				select {

-				case updates <- CertificateUpdate{Err: errors.New("failed to read certificate expiration")}:

-				case <-ctx.Done():

-					log.Info("shutting down certificate renewal routine")

-					return

-				}

-			} else {

-				// If we have an expired certificate, try to renew immediately: the hope that this is a temporary clock skew, or

-				// we can issue our own TLS certs.

-				if validUntil.Before(time.Now()) {

-					log.Warn("the current TLS certificate is expired, so an attempt to renew it will be made immediately")

-					// retry immediately(ish) with exponential backoff

-					retry = expBackoff.Proceed(nil)

-				} else if forceRetry {

-					// A forced renewal was requested, but did not succeed yet.

-					// retry immediately(ish) with exponential backoff

-					retry = expBackoff.Proceed(nil)

-				} else {

-					// Random retry time between 50% and 80% of the total time to expiration

-					retry = calculateRandomExpiry(validFrom, validUntil)

-				}

-			}

-

-			log.WithFields(logrus.Fields{

-				"time": time.Now().Add(retry),

-			}).Debugf("next certificate renewal scheduled for %v from now", retry)

-

-			select {

-			case <-time.After(retry):

-				log.Info("renewing certificate")

-			case <-renew:

-				forceRetry = true

-				log.Info("forced certificate renewal")

-			case <-ctx.Done():

-				log.Info("shutting down certificate renewal routine")

-				return

-			}

-

-			// ignore errors - it will just try again later

-			var certUpdate CertificateUpdate

-			if err := RenewTLSConfigNow(ctx, s, connBroker); err != nil {

-				certUpdate.Err = err

-				expBackoff.Failure(nil, nil)

-			} else {

-				certUpdate.Role = s.ClientTLSCreds.Role()

-				expBackoff = events.NewExponentialBackoff(RenewTLSExponentialBackoff)

-				forceRetry = false

-			}

-

-			select {

-			case updates <- certUpdate:

-			case <-ctx.Done():

-				log.Info("shutting down certificate renewal routine")

-				return

-			}

-		}

-	}()

-

-	return updates

-}

-

 // calculateRandomExpiry returns a random duration between 50% and 80% of the

 // original validity period

 func calculateRandomExpiry(validFrom, validUntil time.Time) time.Duration {

@@ -241,7 +241,7 @@ func (r *rootRotationReconciler) batchUpdateNodes(toUpdate []*api.Node) error {

 	if len(toUpdate) == 0 {

 		return nil

 	}

-	_, err := r.store.Batch(func(batch *store.Batch) error {

+	err := r.store.Batch(func(batch *store.Batch) error {

 		// Directly update the nodes rather than get + update, and ignore version errors.  Since

 		// `rootRotationReconciler` should be hooked up to all node update/delete/create events, we should have

 		// close to the latest versions of all the nodes.  If not, the node will updated later and the

new file mode 100644

@@ -0,0 +1,166 @@

+package ca

+

+import (

+	"sync"

+	"time"

+

+	"github.com/Sirupsen/logrus"

+	"github.com/docker/go-events"

+	"github.com/docker/swarmkit/connectionbroker"

+	"github.com/docker/swarmkit/log"

+	"github.com/pkg/errors"

+	"golang.org/x/net/context"

+)

+

+// RenewTLSExponentialBackoff sets the exponential backoff when trying to renew TLS certificates that have expired

+var RenewTLSExponentialBackoff = events.ExponentialBackoffConfig{

+	Base:   time.Second * 5,

+	Factor: time.Second * 5,

+	Max:    1 * time.Hour,

+}

+

+// TLSRenewer handles renewing TLS certificates, either automatically or upon

+// request.

+type TLSRenewer struct {

+	mu           sync.Mutex

+	s            *SecurityConfig

+	connBroker   *connectionbroker.Broker

+	renew        chan struct{}

+	expectedRole string

+}

+

+// NewTLSRenewer creates a new TLS renewer. It must be started with Start.

+func NewTLSRenewer(s *SecurityConfig, connBroker *connectionbroker.Broker) *TLSRenewer {

+	return &TLSRenewer{

+		s:          s,

+		connBroker: connBroker,

+		renew:      make(chan struct{}, 1),

+	}

+}

+

+// SetExpectedRole sets the expected role. If a renewal is forced, and the role

+// doesn't match this expectation, renewal will be retried with exponential

+// backoff until it does match.

+func (t *TLSRenewer) SetExpectedRole(role string) {

+	t.mu.Lock()

+	t.expectedRole = role

+	t.mu.Unlock()

+}

+

+// Renew causes the TLSRenewer to renew the certificate (nearly) right away,

+// instead of waiting for the next automatic renewal.

+func (t *TLSRenewer) Renew() {

+	select {

+	case t.renew <- struct{}{}:

+	default:

+	}

+}

+

+// Start 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 (t *TLSRenewer) Start(ctx context.Context) <-chan CertificateUpdate {

+	updates := make(chan CertificateUpdate)

+

+	go func() {

+		var (

+			retry      time.Duration

+			forceRetry bool

+		)

+		expBackoff := events.NewExponentialBackoff(RenewTLSExponentialBackoff)

+		defer close(updates)

+		for {

+			ctx = log.WithModule(ctx, "tls")

+			log := log.G(ctx).WithFields(logrus.Fields{

+				"node.id":   t.s.ClientTLSCreds.NodeID(),

+				"node.role": t.s.ClientTLSCreds.Role(),

+			})

+			// Our starting default will be 5 minutes

+			retry = 5 * time.Minute

+

+			// Since the expiration of the certificate is managed remotely we should update our

+			// retry timer on every iteration of this loop.

+			// Retrieve the current certificate expiration information.

+			validFrom, validUntil, err := readCertValidity(t.s.KeyReader())

+			if err != nil {

+				// We failed to read the expiration, let's stick with the starting default

+				log.Errorf("failed to read the expiration of the TLS certificate in: %s", t.s.KeyReader().Target())

+

+				select {

+				case updates <- CertificateUpdate{Err: errors.New("failed to read certificate expiration")}:

+				case <-ctx.Done():

+					log.Info("shutting down certificate renewal routine")

+					return

+				}

+			} else {

+				// If we have an expired certificate, try to renew immediately: the hope that this is a temporary clock skew, or

+				// we can issue our own TLS certs.

+				if validUntil.Before(time.Now()) {

+					log.Warn("the current TLS certificate is expired, so an attempt to renew it will be made immediately")

+					// retry immediately(ish) with exponential backoff

+					retry = expBackoff.Proceed(nil)

+				} else if forceRetry {

+					// A forced renewal was requested, but did not succeed yet.

+					// retry immediately(ish) with exponential backoff

+					retry = expBackoff.Proceed(nil)

+				} else {

+					// Random retry time between 50% and 80% of the total time to expiration

+					retry = calculateRandomExpiry(validFrom, validUntil)

+				}

+			}

+

+			log.WithFields(logrus.Fields{

+				"time": time.Now().Add(retry),

+			}).Debugf("next certificate renewal scheduled for %v from now", retry)

+

+			select {

+			case <-time.After(retry):

+				log.Info("renewing certificate")

+			case <-t.renew:

+				forceRetry = true

+				log.Info("forced certificate renewal")

+

+				// Pause briefly before attempting the renewal,

+				// to give the CA a chance to reconcile the

+				// desired role.

+				select {

+				case <-time.After(500 * time.Millisecond):

+				case <-ctx.Done():

+					log.Info("shutting down certificate renewal routine")

+					return

+				}

+			case <-ctx.Done():

+				log.Info("shutting down certificate renewal routine")

+				return

+			}

+

+			// ignore errors - it will just try again later

+			var certUpdate CertificateUpdate

+			if err := RenewTLSConfigNow(ctx, t.s, t.connBroker); err != nil {

+				certUpdate.Err = err

+				expBackoff.Failure(nil, nil)

+			} else {

+				newRole := t.s.ClientTLSCreds.Role()

+				t.mu.Lock()

+				expectedRole := t.expectedRole

+				t.mu.Unlock()

+				if expectedRole != "" && expectedRole != newRole {

+					expBackoff.Failure(nil, nil)

+					continue

+				}

+

+				certUpdate.Role = newRole

+				expBackoff.Success(nil)

+				forceRetry = false

+			}

+

+			select {

+			case updates <- certUpdate:

+			case <-ctx.Done():

+				log.Info("shutting down certificate renewal routine")

+				return

+			}

+		}

+	}()

+

+	return updates

+}

@@ -580,6 +580,7 @@ func (s *Server) UpdateRootCA(ctx context.Context, cluster *api.Cluster) error {

 	s.secConfigMu.Lock()

 	defer s.secConfigMu.Unlock()

+	firstSeenCluster := s.lastSeenClusterRootCA == nil && s.lastSeenExternalCAs == nil

 	rootCAChanged := len(rCA.CACert) != 0 && !equality.RootCAEqualStable(s.lastSeenClusterRootCA, rCA)

 	externalCAChanged := !equality.ExternalCAsEqualStable(s.lastSeenExternalCAs, cluster.Spec.CAConfig.ExternalCAs)

 	logger := log.G(ctx).WithFields(logrus.Fields{

@@ -588,7 +589,11 @@ func (s *Server) UpdateRootCA(ctx context.Context, cluster *api.Cluster) error {

 	})

 	if rootCAChanged {

-		logger.Debug("Updating security config due to change in cluster Root CA")

+		setOrUpdate := "set"

+		if !firstSeenCluster {

+			logger.Debug("Updating security config due to change in cluster Root CA")

+			setOrUpdate = "updated"

+		}

 		expiry := DefaultNodeCertExpiration

 		if cluster.Spec.CAConfig.NodeCertExpiry != nil {

 			// NodeCertExpiry exists, let's try to parse the duration out of it

@@ -636,14 +641,16 @@ func (s *Server) UpdateRootCA(ctx context.Context, cluster *api.Cluster) error {

 			return errors.Wrap(err, "updating Root CA failed")

 		}

 		// only update the server cache if we've successfully updated the root CA

-		logger.Debug("Root CA updated successfully")

+		logger.Debugf("Root CA %s successfully", setOrUpdate)

 		s.lastSeenClusterRootCA = rCA

 	}

 	// we want to update if the external CA changed, or if the root CA changed because the root CA could affect what

 	// certificate for external CAs we want to filter by

 	if rootCAChanged || externalCAChanged {

-		logger.Debug("Updating security config due to change in cluster Root CA or cluster spec")

+		if !firstSeenCluster {

+			logger.Debug("Updating security config external CA URLs due to change in cluster Root CA or cluster spec")

+		}

 		wantedExternalCACert := rCA.CACert // we want to only add external CA URLs that use this cert

 		if rCA.RootRotation != nil {

 			// we're rotating to a new root, so we only want external CAs with the new root cert

@@ -1,6 +1,6 @@

-// Package identity provides functionality for generating and manager

-// identifiers within swarm. This includes entity identification, such as that

-// of Service, Task and Network but also cryptographically-secure Node identity.

+// Package identity provides functionality for generating and managing

+// identifiers within a swarm. This includes entity identification, such as for

+// Services, Tasks and Networks but also cryptographically-secure Node identities.

 //

 // Random Identifiers

 //

@@ -8,10 +8,9 @@

 // 128 bit numbers encoded in Base36. This method is preferred over UUID4 since

 // it requires less storage and leverages the full 128 bits of entropy.

 //

-// Generating an identifier is simple. Simply call the `NewID` function, check

-// the error and proceed:

+// Generating an identifier is simple. Simply call the `NewID` function:

 //

-// 	id, err := NewID()

-// 	if err != nil { /* ... handle it, please ... */ }

+// 	id := NewID()

 //

+// If an error occurs while generating the ID, it will panic.

 package identity

@@ -3,16 +3,16 @@

 // manages a set of independent allocator processes which can mostly

 // execute concurrently with only a minimal need for coordination.

 //

-// One of the instances where it needs coordination is when to move a

-// task to ALLOCATED state. Since a task can move to ALLOCATED state

-// only when all task allocators have completed their service of

-// allocation, they all have to agree on that. The way this achieved

-// in `allocator` is by creating a `taskBallot` to which all task

-// allocators register themselves as mandatory voters. For each task

-// that needs allocation, each allocator independently votes to indicate

-// the completion of their allocation. Once all registered voters have

-// voted then the task is moved to ALLOCATED state.

+// One of the instances where it needs coordination is when deciding to

+// move a task to the PENDING state. Since a task can move to the

+// PENDING state only when all the task allocators have completed,

+// they must cooperate. The way `allocator` achieves this is by creating

+// a `taskBallot` to which all task allocators register themselves as

+// mandatory voters. For each task that needs allocation, each allocator

+// independently votes to indicate the completion of their allocation.

+// Once all registered voters have voted then the task is moved to the

+// PENDING state.

 //

-// Other than the coordination needed for task ALLOCATED state, all

+// Other than the coordination needed for task PENDING state, all

 // the allocators function fairly independently.

 package allocator

@@ -95,7 +95,7 @@ func (a *Allocator) doNetworkInit(ctx context.Context) (err error) {

 		if !na.IsAllocated(nc.ingressNetwork) {

 			if err := a.allocateNetwork(ctx, nc.ingressNetwork); err != nil {

 				log.G(ctx).WithError(err).Error("failed allocating ingress network during init")

-			} else if _, err := a.store.Batch(func(batch *store.Batch) error {

+			} else if err := a.store.Batch(func(batch *store.Batch) error {

 				if err := a.commitAllocatedNetwork(ctx, batch, nc.ingressNetwork); err != nil {

 					log.G(ctx).WithError(err).Error("failed committing allocation of ingress network during init")

 				}

@@ -134,7 +134,7 @@ func (a *Allocator) doNetworkInit(ctx context.Context) (err error) {

 		allocatedNetworks = append(allocatedNetworks, n)

 	}

-	if _, err := a.store.Batch(func(batch *store.Batch) error {

+	if err := a.store.Batch(func(batch *store.Batch) error {

 		for _, n := range allocatedNetworks {

 			if err := a.commitAllocatedNetwork(ctx, batch, n); err != nil {

 				log.G(ctx).WithError(err).Errorf("failed committing allocation of network %s during init", n.ID)

@@ -164,7 +164,7 @@ func (a *Allocator) doNetworkInit(ctx context.Context) (err error) {

 	var allocatedServices []*api.Service

 	for _, s := range services {

-		if nc.nwkAllocator.IsServiceAllocated(s, networkallocator.OnInit) {

+		if !nc.nwkAllocator.ServiceNeedsAllocation(s, networkallocator.OnInit) {

 			continue

 		}

@@ -175,7 +175,7 @@ func (a *Allocator) doNetworkInit(ctx context.Context) (err error) {

 		allocatedServices = append(allocatedServices, s)

 	}

-	if _, err := a.store.Batch(func(batch *store.Batch) error {

+	if err := a.store.Batch(func(batch *store.Batch) error {

 		for _, s := range allocatedServices {

 			if err := a.commitAllocatedService(ctx, batch, s); err != nil {

 				log.G(ctx).WithError(err).Errorf("failed committing allocation of service %s during init", s.ID)

@@ -239,7 +239,7 @@ func (a *Allocator) doNetworkInit(ctx context.Context) (err error) {

 		}

 	}

-	if _, err := a.store.Batch(func(batch *store.Batch) error {

+	if err := a.store.Batch(func(batch *store.Batch) error {

 		for _, t := range allocatedTasks {

 			if err := a.commitAllocatedTask(ctx, batch, t); err != nil {

 				log.G(ctx).WithError(err).Errorf("failed committing allocation of task %s during init", t.ID)

@@ -275,7 +275,7 @@ func (a *Allocator) doNetworkAlloc(ctx context.Context, ev events.Event) {

 			break

 		}

-		if _, err := a.store.Batch(func(batch *store.Batch) error {

+		if err := a.store.Batch(func(batch *store.Batch) error {

 			return a.commitAllocatedNetwork(ctx, batch, n)

 		}); err != nil {

 			log.G(ctx).WithError(err).Errorf("Failed to commit allocation for network %s", n.ID)

@@ -317,7 +317,7 @@ func (a *Allocator) doNetworkAlloc(ctx context.Context, ev events.Event) {

 			break

 		}

-		if nc.nwkAllocator.IsServiceAllocated(s) {

+		if !nc.nwkAllocator.ServiceNeedsAllocation(s) {

 			break

 		}

@@ -326,7 +326,7 @@ func (a *Allocator) doNetworkAlloc(ctx context.Context, ev events.Event) {

 			break

 		}

-		if _, err := a.store.Batch(func(batch *store.Batch) error {

+		if err := a.store.Batch(func(batch *store.Batch) error {

 			return a.commitAllocatedService(ctx, batch, s)

 		}); err != nil {

 			log.G(ctx).WithError(err).Errorf("Failed to commit allocation for service %s", s.ID)

@@ -345,8 +345,8 @@ func (a *Allocator) doNetworkAlloc(ctx context.Context, ev events.Event) {

 			break

 		}

-		if nc.nwkAllocator.IsServiceAllocated(s) {

-			if nc.nwkAllocator.PortsAllocatedInHostPublishMode(s) {

+		if !nc.nwkAllocator.ServiceNeedsAllocation(s) {

+			if !nc.nwkAllocator.HostPublishPortsNeedUpdate(s) {

 				break

 			}

 			updatePortsInHostPublishMode(s)

@@ -357,7 +357,7 @@ func (a *Allocator) doNetworkAlloc(ctx context.Context, ev events.Event) {

 			}

 		}

-		if _, err := a.store.Batch(func(batch *store.Batch) error {

+		if err := a.store.Batch(func(batch *store.Batch) error {

 			return a.commitAllocatedService(ctx, batch, s)

 		}); err != nil {

 			log.G(ctx).WithError(err).Errorf("Failed to commit allocation during update for service %s", s.ID)

@@ -447,7 +447,7 @@ func (a *Allocator) doNodeAlloc(ctx context.Context, ev events.Event) {

 			return

 		}

-		if _, err := a.store.Batch(func(batch *store.Batch) error {

+		if err := a.store.Batch(func(batch *store.Batch) error {

 			return a.commitAllocatedNode(ctx, batch, node)

 		}); err != nil {

 			log.G(ctx).WithError(err).Errorf("Failed to commit allocation of network resources for node %s", node.ID)

@@ -489,7 +489,7 @@ func (a *Allocator) allocateNodes(ctx context.Context) error {

 		allocatedNodes = append(allocatedNodes, node)

 	}

-	if _, err := a.store.Batch(func(batch *store.Batch) error {

+	if err := a.store.Batch(func(batch *store.Batch) error {

 		for _, node := range allocatedNodes {

 			if err := a.commitAllocatedNode(ctx, batch, node); err != nil {

 				log.G(ctx).WithError(err).Errorf("Failed to commit allocation of network resources for node %s", node.ID)

@@ -523,7 +523,7 @@ func (a *Allocator) deallocateNodes(ctx context.Context) error {

 				log.G(ctx).WithError(err).Errorf("Failed freeing network resources for node %s", node.ID)

 			}

 			node.Attachment = nil

-			if _, err := a.store.Batch(func(batch *store.Batch) error {

+			if err := a.store.Batch(func(batch *store.Batch) error {

 				return a.commitAllocatedNode(ctx, batch, node)

 			}); err != nil {

 				log.G(ctx).WithError(err).Errorf("Failed to commit deallocation of network resources for node %s", node.ID)

@@ -544,7 +544,7 @@ func taskReadyForNetworkVote(t *api.Task, s *api.Service, nc *networkContext) bo

 	// network configured or service endpoints have been

 	// allocated.

 	return (len(t.Networks) == 0 || nc.nwkAllocator.IsTaskAllocated(t)) &&

-		(s == nil || nc.nwkAllocator.IsServiceAllocated(s))

+		(s == nil || !nc.nwkAllocator.ServiceNeedsAllocation(s))

 }

 func taskUpdateNetworks(t *api.Task, networks []*api.NetworkAttachment) {

@@ -732,28 +732,29 @@ func (a *Allocator) commitAllocatedNode(ctx context.Context, batch *store.Batch,

 // so that the service allocation invoked on this new service object will trigger the deallocation

 // of any old publish mode port and allocation of any new one.

 func updatePortsInHostPublishMode(s *api.Service) {

+	// First, remove all host-mode ports from s.Endpoint.Ports

 	if s.Endpoint != nil {

 		var portConfigs []*api.PortConfig

 		for _, portConfig := range s.Endpoint.Ports {

-			if portConfig.PublishMode == api.PublishModeIngress {

+			if portConfig.PublishMode != api.PublishModeHost {

 				portConfigs = append(portConfigs, portConfig)

 			}

 		}

 		s.Endpoint.Ports = portConfigs

 	}

+	// Add back all host-mode ports

 	if s.Spec.Endpoint != nil {

 		if s.Endpoint == nil {

 			s.Endpoint = &api.Endpoint{}

 		}

 		for _, portConfig := range s.Spec.Endpoint.Ports {

-			if portConfig.PublishMode == api.PublishModeIngress {

-				continue

+			if portConfig.PublishMode == api.PublishModeHost {

+				s.Endpoint.Ports = append(s.Endpoint.Ports, portConfig.Copy())

 			}

-			s.Endpoint.Ports = append(s.Endpoint.Ports, portConfig.Copy())

 		}

-		s.Endpoint.Spec = s.Spec.Endpoint.Copy()

 	}

+	s.Endpoint.Spec = s.Spec.Endpoint.Copy()

 }

 func (a *Allocator) allocateService(ctx context.Context, s *api.Service) error {

@@ -886,7 +887,7 @@ func (a *Allocator) allocateTask(ctx context.Context, t *api.Task) (err error) {

 					return

 				}

-				if !nc.nwkAllocator.IsServiceAllocated(s) {

+				if nc.nwkAllocator.ServiceNeedsAllocation(s) {

 					err = fmt.Errorf("service %s to which this task %s belongs has pending allocations", s.ID, t.ID)

 					return

 				}

@@ -977,22 +978,25 @@ func (a *Allocator) procUnallocatedNetworks(ctx context.Context) {

 		return

 	}

-	committed, err := a.store.Batch(func(batch *store.Batch) error {

+	err := a.store.Batch(func(batch *store.Batch) error {

 		for _, n := range allocatedNetworks {

 			if err := a.commitAllocatedNetwork(ctx, batch, n); err != nil {

 				log.G(ctx).WithError(err).Debugf("Failed to commit allocation of unallocated network %s", n.ID)

 				continue

 			}

+			delete(nc.unallocatedNetworks, n.ID)

 		}

 		return nil

 	})

 	if err != nil {

 		log.G(ctx).WithError(err).Error("Failed to commit allocation of unallocated networks")

-	}

-

-	for _, n := range allocatedNetworks[:committed] {

-		delete(nc.unallocatedNetworks, n.ID)

+		// We optimistically removed these from nc.unallocatedNetworks

+		// above in anticipation of successfully committing the batch,

+		// but since the transaction has failed, we requeue them here.

+		for _, n := range allocatedNetworks {

+			nc.unallocatedNetworks[n.ID] = n

+		}

 	}

 }

@@ -1000,7 +1004,7 @@ func (a *Allocator) procUnallocatedServices(ctx context.Context) {

 	nc := a.netCtx

 	var allocatedServices []*api.Service

 	for _, s := range nc.unallocatedServices {

-		if !nc.nwkAllocator.IsServiceAllocated(s) {

+		if nc.nwkAllocator.ServiceNeedsAllocation(s) {

 			if err := a.allocateService(ctx, s); err != nil {

 				log.G(ctx).WithError(err).Debugf("Failed allocation of unallocated service %s", s.ID)

 				continue

@@ -1013,22 +1017,25 @@ func (a *Allocator) procUnallocatedServices(ctx context.Context) {

 		return

 	}

-	committed, err := a.store.Batch(func(batch *store.Batch) error {

+	err := a.store.Batch(func(batch *store.Batch) error {

 		for _, s := range allocatedServices {

 			if err := a.commitAllocatedService(ctx, batch, s); err != nil {

 				log.G(ctx).WithError(err).Debugf("Failed to commit allocation of unallocated service %s", s.ID)

 				continue

 			}

+			delete(nc.unallocatedServices, s.ID)

 		}

 		return nil

 	})

 	if err != nil {

 		log.G(ctx).WithError(err).Error("Failed to commit allocation of unallocated services")

-	}

-

-	for _, s := range allocatedServices[:committed] {

-		delete(nc.unallocatedServices, s.ID)

+		// We optimistically removed these from nc.unallocatedServices

+		// above in anticipation of successfully committing the batch,

+		// but since the transaction has failed, we requeue them here.

+		for _, s := range allocatedServices {

+			nc.unallocatedServices[s.ID] = s

+		}

 	}

 }

@@ -1058,14 +1065,14 @@ func (a *Allocator) procTasksNetwork(ctx context.Context, onRetry bool) {

 		return

 	}

-	committed, err := a.store.Batch(func(batch *store.Batch) error {

+	err := a.store.Batch(func(batch *store.Batch) error {

 		for _, t := range allocatedTasks {

 			err := a.commitAllocatedTask(ctx, batch, t)

-

 			if err != nil {

 				log.G(ctx).WithError(err).Error("task allocation commit failure")

 				continue

 			}

+			delete(toAllocate, t.ID)

 		}

 		return nil

@@ -1073,10 +1080,12 @@ func (a *Allocator) procTasksNetwork(ctx context.Context, onRetry bool) {

 	if err != nil {

 		log.G(ctx).WithError(err).Error("failed a store batch operation while processing tasks")

-	}

-

-	for _, t := range allocatedTasks[:committed] {

-		delete(toAllocate, t.ID)

+		// We optimistically removed these from toAllocate above in

+		// anticipation of successfully committing the batch, but since

+		// the transaction has failed, we requeue them here.

+		for _, t := range allocatedTasks {

+			toAllocate[t.ID] = t

+		}

 	}

 }

@@ -1089,12 +1098,7 @@ func updateTaskStatus(t *api.Task, newStatus api.TaskState, message string) {

 // IsIngressNetwork returns whether the passed network is an ingress network.

 func IsIngressNetwork(nw *api.Network) bool {

-	if nw.Spec.Ingress {

-		return true

-	}

-	// Check if legacy defined ingress network

-	_, ok := nw.Spec.Annotations.Labels["com.docker.swarm.internal"]

-	return ok && nw.Spec.Annotations.Name == "ingress"

+	return networkallocator.IsIngressNetwork(nw)

 }

 // GetIngressNetwork fetches the ingress network from store.

@@ -153,7 +153,7 @@ func (na *NetworkAllocator) Deallocate(n *api.Network) error {

 // IP and ports needed by the service.

 func (na *NetworkAllocator) ServiceAllocate(s *api.Service) (err error) {

 	if err = na.portAllocator.serviceAllocatePorts(s); err != nil {

-		return

+		return err

 	}

 	defer func() {

 		if err != nil {

@@ -169,54 +169,74 @@ func (na *NetworkAllocator) ServiceAllocate(s *api.Service) (err error) {

 	// If ResolutionMode is DNSRR do not try allocating VIPs, but

 	// free any VIP from previous state.

 	if s.Spec.Endpoint != nil && s.Spec.Endpoint.Mode == api.ResolutionModeDNSRoundRobin {

-		if s.Endpoint != nil {

-			for _, vip := range s.Endpoint.VirtualIPs {

-				if err := na.deallocateVIP(vip); err != nil {

-					// don't bail here, deallocate as many as possible.

-					log.L.WithError(err).

-						WithField("vip.network", vip.NetworkID).

-						WithField("vip.addr", vip.Addr).Error("error deallocating vip")

-				}

+		for _, vip := range s.Endpoint.VirtualIPs {

+			if err := na.deallocateVIP(vip); err != nil {

+				// don't bail here, deallocate as many as possible.

+				log.L.WithError(err).

+					WithField("vip.network", vip.NetworkID).

+					WithField("vip.addr", vip.Addr).Error("error deallocating vip")

 			}

-

-			s.Endpoint.VirtualIPs = nil

 		}

+		s.Endpoint.VirtualIPs = nil

+

 		delete(na.services, s.ID)

-		return

+		return nil

 	}

-	// First allocate VIPs for all the pre-populated endpoint attachments

+	specNetworks := serviceNetworks(s)

+

+	// Allocate VIPs for all the pre-populated endpoint attachments

+	eVIPs := s.Endpoint.VirtualIPs[:0]

+

+vipLoop:

 	for _, eAttach := range s.Endpoint.VirtualIPs {

-		if err = na.allocateVIP(eAttach); err != nil {

-			return

-		}

-	}

+		if na.IsVIPOnIngressNetwork(eAttach) {

+			if err = na.allocateVIP(eAttach); err != nil {

+				return err

+			}

+			eVIPs = append(eVIPs, eAttach)

+			continue vipLoop

-	// Always prefer NetworkAttachmentConfig in the TaskSpec

-	specNetworks := s.Spec.Task.Networks

-	if len(specNetworks) == 0 && s != nil && len(s.Spec.Networks) != 0 {

-		specNetworks = s.Spec.Networks

+		}

+		for _, nAttach := range specNetworks {

+			if nAttach.Target == eAttach.NetworkID {

+				if err = na.allocateVIP(eAttach); err != nil {

+					return err

+				}

+				eVIPs = append(eVIPs, eAttach)

+				continue vipLoop

+			}

+		}

+		// If the network of the VIP is not part of the service spec,

+		// deallocate the vip

+		na.deallocateVIP(eAttach)

 	}

-outer:

+networkLoop:

 	for _, nAttach := range specNetworks {

 		for _, vip := range s.Endpoint.VirtualIPs {

 			if vip.NetworkID == nAttach.Target {

-				continue outer

+				continue networkLoop

 			}

 		}

 		vip := &api.Endpoint_VirtualIP{NetworkID: nAttach.Target}

 		if err = na.allocateVIP(vip); err != nil {

-			return

+			return err

 		}

-		s.Endpoint.VirtualIPs = append(s.Endpoint.VirtualIPs, vip)

+		eVIPs = append(eVIPs, vip)

+	}

+

+	if len(eVIPs) > 0 {

+		na.services[s.ID] = struct{}{}