@@ -362,9 +362,8 @@ func (daemon *Daemon) GetNetworkDriverList() []string {

 		return nil

 	}

-	// TODO: Replace this with proper libnetwork API

-	pluginList := []string{"overlay"}

-	pluginMap := map[string]bool{"overlay": true}

+	pluginList := daemon.netController.BuiltinDrivers()

+	pluginMap := make(map[string]bool)

 	networks := daemon.netController.Networks()

@@ -146,7 +146,7 @@ clone git github.com/docker/docker-credential-helpers v0.3.0

 clone git github.com/docker/containerd 52ef1ceb4b660c42cf4ea9013180a5663968d4c7

 # cluster

-clone git github.com/docker/swarmkit 7e63bdefb94e5bea2641e8bdebae2cfa61a0ed44

+clone git github.com/docker/swarmkit 1fed8d2a2ccd2a9b6d6fb864d4ad3461fc6dc3eb

 clone git github.com/golang/mock bd3c8e81be01eef76d4b503f5e687d2d1354d2d9

 clone git github.com/gogo/protobuf v0.3

 clone git github.com/cloudflare/cfssl 7fb22c8cba7ecaf98e4082d22d65800cf45e042a

@@ -2,6 +2,7 @@ package main

 import (

 	"fmt"

+	"net/http/httptest"

 	"os"

 	"path/filepath"

 	"sync"

@@ -240,6 +241,7 @@ func init() {

 }

 type DockerSwarmSuite struct {

+	server      *httptest.Server

 	ds          *DockerSuite

 	daemons     []*SwarmDaemon

 	daemonsLock sync.Mutex // protect access to daemons

@@ -3,13 +3,22 @@

 package main

 import (

+	"encoding/json"

+	"fmt"

 	"io/ioutil"

+	"net/http"

+	"net/http/httptest"

+	"os"

 	"strings"

 	"time"

 	"github.com/docker/docker/api/types/swarm"

 	"github.com/docker/docker/pkg/integration/checker"

+	"github.com/docker/libnetwork/driverapi"

+	"github.com/docker/libnetwork/ipamapi"

+	remoteipam "github.com/docker/libnetwork/ipams/remote/api"

 	"github.com/go-check/check"

+	"github.com/vishvananda/netlink"

 )

 func (s *DockerSwarmSuite) TestSwarmUpdate(c *check.C) {

@@ -364,3 +373,198 @@ func (s *DockerSwarmSuite) TestPsListContainersFilterIsTask(c *check.C) {

 	c.Assert(lines, checker.HasLen, 1)

 	c.Assert(lines[0], checker.Not(checker.Equals), bareID, check.Commentf("Expected not %s, but got it for is-task label, output %q", bareID, out))

 }

+

+const globalNetworkPlugin = "global-network-plugin"

+const globalIPAMPlugin = "global-ipam-plugin"

+

+func (s *DockerSwarmSuite) SetUpSuite(c *check.C) {

+	mux := http.NewServeMux()

+	s.server = httptest.NewServer(mux)

+	c.Assert(s.server, check.NotNil, check.Commentf("Failed to start an HTTP Server"))

+	setupRemoteGlobalNetworkPlugin(c, mux, s.server.URL, globalNetworkPlugin, globalIPAMPlugin)

+}

+

+func setupRemoteGlobalNetworkPlugin(c *check.C, mux *http.ServeMux, url, netDrv, ipamDrv string) {

+

+	mux.HandleFunc("/Plugin.Activate", func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, `{"Implements": ["%s", "%s"]}`, driverapi.NetworkPluginEndpointType, ipamapi.PluginEndpointType)

+	})

+

+	// Network driver implementation

+	mux.HandleFunc(fmt.Sprintf("/%s.GetCapabilities", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, `{"Scope":"global"}`)

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.AllocateNetwork", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		err := json.NewDecoder(r.Body).Decode(&remoteDriverNetworkRequest)

+		if err != nil {

+			http.Error(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest)

+			return

+		}

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, "null")

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.FreeNetwork", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, "null")

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.CreateNetwork", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		err := json.NewDecoder(r.Body).Decode(&remoteDriverNetworkRequest)

+		if err != nil {

+			http.Error(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest)

+			return

+		}

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, "null")

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.DeleteNetwork", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, "null")

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.CreateEndpoint", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, `{"Interface":{"MacAddress":"a0:b1:c2:d3:e4:f5"}}`)

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.Join", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+

+		veth := &netlink.Veth{

+			LinkAttrs: netlink.LinkAttrs{Name: "randomIfName", TxQLen: 0}, PeerName: "cnt0"}

+		if err := netlink.LinkAdd(veth); err != nil {

+			fmt.Fprintf(w, `{"Error":"failed to add veth pair: `+err.Error()+`"}`)

+		} else {

+			fmt.Fprintf(w, `{"InterfaceName":{ "SrcName":"cnt0", "DstPrefix":"veth"}}`)

+		}

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.Leave", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, "null")

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.DeleteEndpoint", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		if link, err := netlink.LinkByName("cnt0"); err == nil {

+			netlink.LinkDel(link)

+		}

+		fmt.Fprintf(w, "null")

+	})

+

+	// IPAM Driver implementation

+	var (

+		poolRequest       remoteipam.RequestPoolRequest

+		poolReleaseReq    remoteipam.ReleasePoolRequest

+		addressRequest    remoteipam.RequestAddressRequest

+		addressReleaseReq remoteipam.ReleaseAddressRequest

+		lAS               = "localAS"

+		gAS               = "globalAS"

+		pool              = "172.28.0.0/16"

+		poolID            = lAS + "/" + pool

+		gw                = "172.28.255.254/16"

+	)

+

+	mux.HandleFunc(fmt.Sprintf("/%s.GetDefaultAddressSpaces", ipamapi.PluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		fmt.Fprintf(w, `{"LocalDefaultAddressSpace":"`+lAS+`", "GlobalDefaultAddressSpace": "`+gAS+`"}`)

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.RequestPool", ipamapi.PluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		err := json.NewDecoder(r.Body).Decode(&poolRequest)

+		if err != nil {

+			http.Error(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest)

+			return

+		}

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		if poolRequest.AddressSpace != lAS && poolRequest.AddressSpace != gAS {

+			fmt.Fprintf(w, `{"Error":"Unknown address space in pool request: `+poolRequest.AddressSpace+`"}`)

+		} else if poolRequest.Pool != "" && poolRequest.Pool != pool {

+			fmt.Fprintf(w, `{"Error":"Cannot handle explicit pool requests yet"}`)

+		} else {

+			fmt.Fprintf(w, `{"PoolID":"`+poolID+`", "Pool":"`+pool+`"}`)

+		}

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.RequestAddress", ipamapi.PluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		err := json.NewDecoder(r.Body).Decode(&addressRequest)

+		if err != nil {

+			http.Error(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest)

+			return

+		}

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		// make sure libnetwork is now querying on the expected pool id

+		if addressRequest.PoolID != poolID {

+			fmt.Fprintf(w, `{"Error":"unknown pool id"}`)

+		} else if addressRequest.Address != "" {

+			fmt.Fprintf(w, `{"Error":"Cannot handle explicit address requests yet"}`)

+		} else {

+			fmt.Fprintf(w, `{"Address":"`+gw+`"}`)

+		}

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.ReleaseAddress", ipamapi.PluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		err := json.NewDecoder(r.Body).Decode(&addressReleaseReq)

+		if err != nil {

+			http.Error(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest)

+			return

+		}

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		// make sure libnetwork is now asking to release the expected address from the expected poolid

+		if addressRequest.PoolID != poolID {

+			fmt.Fprintf(w, `{"Error":"unknown pool id"}`)

+		} else if addressReleaseReq.Address != gw {

+			fmt.Fprintf(w, `{"Error":"unknown address"}`)

+		} else {

+			fmt.Fprintf(w, "null")

+		}

+	})

+

+	mux.HandleFunc(fmt.Sprintf("/%s.ReleasePool", ipamapi.PluginEndpointType), func(w http.ResponseWriter, r *http.Request) {

+		err := json.NewDecoder(r.Body).Decode(&poolReleaseReq)

+		if err != nil {

+			http.Error(w, "Unable to decode JSON payload: "+err.Error(), http.StatusBadRequest)

+			return

+		}

+		w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")

+		// make sure libnetwork is now asking to release the expected poolid

+		if addressRequest.PoolID != poolID {

+			fmt.Fprintf(w, `{"Error":"unknown pool id"}`)

+		} else {

+			fmt.Fprintf(w, "null")

+		}

+	})

+

+	err := os.MkdirAll("/etc/docker/plugins", 0755)

+	c.Assert(err, checker.IsNil)

+

+	fileName := fmt.Sprintf("/etc/docker/plugins/%s.spec", netDrv)

+	err = ioutil.WriteFile(fileName, []byte(url), 0644)

+	c.Assert(err, checker.IsNil)

+

+	ipamFileName := fmt.Sprintf("/etc/docker/plugins/%s.spec", ipamDrv)

+	err = ioutil.WriteFile(ipamFileName, []byte(url), 0644)

+	c.Assert(err, checker.IsNil)

+}

+

+func (s *DockerSwarmSuite) TestSwarmNetworkPlugin(c *check.C) {

+	d := s.AddDaemon(c, true, true)

+

+	out, err := d.Cmd("network", "create", "-d", globalNetworkPlugin, "foo")

+	c.Assert(err, checker.IsNil)

+	c.Assert(strings.TrimSpace(out), checker.Not(checker.Equals), "")

+

+	name := "top"

+	out, err = d.Cmd("service", "create", "--name", name, "--network", "foo", "busybox", "top")

+	c.Assert(err, checker.IsNil)

+	c.Assert(strings.TrimSpace(out), checker.Not(checker.Equals), "")

+

+	out, err = d.Cmd("service", "inspect", "--format", "{{range .Spec.Networks}}{{.Target}}{{end}}", name)

+	c.Assert(err, checker.IsNil)

+	c.Assert(strings.TrimSpace(out), checker.Equals, "foo")

+}

@@ -43,7 +43,7 @@ type plugins struct {

 var (

 	storage          = plugins{plugins: make(map[string]*Plugin)}

-	extpointHandlers = make(map[string]func(string, *Client))

+	extpointHandlers = make(map[string][]func(string, *Client))

 )

 // Manifest lists what a plugin implements.

@@ -129,11 +129,13 @@ func (p *Plugin) activateWithLock() error {

 	p.Manifest = m

 	for _, iface := range m.Implements {

-		handler, handled := extpointHandlers[iface]

+		handlers, handled := extpointHandlers[iface]

 		if !handled {

 			continue

 		}

-		handler(p.name, p.client)

+		for _, handler := range handlers {

+			handler(p.name, p.client)

+		}

 	}

 	return nil

 }

@@ -226,7 +228,16 @@ func Get(name, imp string) (*Plugin, error) {

 // Handle adds the specified function to the extpointHandlers.

 func Handle(iface string, fn func(string, *Client)) {

-	extpointHandlers[iface] = fn

+	handlers, ok := extpointHandlers[iface]

+	if !ok {

+		handlers = []func(string, *Client){}

+	}

+

+	handlers = append(handlers, fn)

+	extpointHandlers[iface] = handlers

+	for _, p := range storage.plugins {

+		p.activated = false

+	}

 }

 // GetAll returns all the plugins for the specified implementation

@@ -15,7 +15,7 @@ type Store struct {

 	/* handlers are necessary for transition path of legacy plugins

 	 * to the new model. Legacy plugins use Handle() for registering an

 	 * activation callback.*/

-	handlers map[string]func(string, *plugins.Client)

+	handlers map[string][]func(string, *plugins.Client)

 	nameToID map[string]string

 	plugindb string

 }

@@ -24,7 +24,7 @@ type Store struct {

 func NewStore(libRoot string) *Store {

 	return &Store{

 		plugins:  make(map[string]*v2.Plugin),

-		handlers: make(map[string]func(string, *plugins.Client)),

+		handlers: make(map[string][]func(string, *plugins.Client)),

 		nameToID: make(map[string]string),

 		plugindb: filepath.Join(libRoot, "plugins", "plugins.json"),

 	}

@@ -218,7 +218,12 @@ func (ps *Store) Handle(capability string, callback func(string, *plugins.Client

 	// Register callback with new plugin model.

 	ps.Lock()

-	ps.handlers[pluginType] = callback

+	handlers, ok := ps.handlers[pluginType]

+	if !ok {

+		handlers = []func(string, *plugins.Client){}

+	}

+	handlers = append(handlers, callback)

+	ps.handlers[pluginType] = handlers

 	ps.Unlock()

 	// Register callback with legacy plugin model.

@@ -230,7 +235,7 @@ func (ps *Store) Handle(capability string, callback func(string, *plugins.Client

 // CallHandler calls the registered callback. It is invoked during plugin enable.

 func (ps *Store) CallHandler(p *v2.Plugin) {

 	for _, typ := range p.GetTypes() {

-		if handler := ps.handlers[typ.String()]; handler != nil {

+		for _, handler := range ps.handlers[typ.String()] {

 			handler(p.Name(), p.Client())

 		}

 	}

@@ -171,11 +171,12 @@ func (a *Agent) run(ctx context.Context) {

 		case msg := <-session.assignments:

 			switch msg.Type {

 			case api.AssignmentsMessage_COMPLETE:

-				if err := a.worker.AssignTasks(ctx, msg.UpdateTasks); err != nil {

+				// Need to assign secrets before tasks, because tasks might depend on new secrets

+				if err := a.worker.Assign(ctx, msg.Changes); err != nil {

 					log.G(ctx).WithError(err).Error("failed to synchronize worker assignments")

 				}

 			case api.AssignmentsMessage_INCREMENTAL:

-				if err := a.worker.UpdateTasks(ctx, msg.UpdateTasks, msg.RemoveTasks); err != nil {

+				if err := a.worker.Update(ctx, msg.Changes); err != nil {

 					log.G(ctx).WithError(err).Error("failed to update worker assignments")

 				}

 			}

@@ -101,6 +101,21 @@ type Node struct {

 	roleChangeReq        chan api.NodeRole // used to send role updates from the dispatcher api on promotion/demotion

 }

+// RemoteAPIAddr returns address on which remote manager api listens.

+// Returns nil if node is not manager.

+func (n *Node) RemoteAPIAddr() (string, error) {

+	n.RLock()

+	defer n.RUnlock()

+	if n.manager == nil {

+		return "", errors.Errorf("node is not manager")

+	}

+	addr := n.manager.Addr()

+	if addr == nil {

+		return "", errors.Errorf("manager addr is not set")

+	}

+	return addr.String(), nil

+}

+

 // NewNode returns new Node instance.

 func NewNode(c *NodeConfig) (*Node, error) {

 	if err := os.MkdirAll(c.StateDir, 0700); err != nil {

@@ -627,7 +642,12 @@ func (n *Node) runManager(ctx context.Context, securityConfig *ca.SecurityConfig

 			go func(ready chan struct{}) {

 				select {

 				case <-ready:

-					n.remotes.Observe(api.Peer{NodeID: n.NodeID(), Addr: n.config.ListenRemoteAPI}, remotes.DefaultObservationWeight)

+					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)

new file mode 100644

@@ -0,0 +1,53 @@

+package agent

+

+import (

+	"sync"

+

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

+)

+

+// secrets is a map that keeps all the currenty available secrets to the agent

+// mapped by secret ID

+type secrets struct {

+	mu sync.RWMutex

+	m  map[string]api.Secret

+}

+

+func newSecrets() *secrets {

+	return &secrets{

+		m: make(map[string]api.Secret),

+	}

+}

+

+// Get returns a secret by ID.  If the secret doesn't exist, returns nil.

+func (s *secrets) Get(secretID string) api.Secret {

+	s.mu.RLock()

+	defer s.mu.RUnlock()

+	return s.m[secretID]

+}

+

+// Add adds one or more secrets to the secret map

+func (s *secrets) Add(secrets ...api.Secret) {

+	s.mu.Lock()

+	defer s.mu.Unlock()

+	for _, secret := range secrets {

+		s.m[secret.ID] = secret

+	}

+}

+

+// Remove removes one or more secrets by ID from the secret map.  Succeeds

+// whether or not the given IDs are in the map.

+func (s *secrets) Remove(secrets []string) {

+	s.mu.Lock()

+	defer s.mu.Unlock()

+	for _, secret := range secrets {

+		delete(s.m, secret)

+	}

+}

+

+// Reset removes all the secrets

+func (s *secrets) Reset() {

+	s.mu.Lock()

+	defer s.mu.Unlock()

+	s.m = make(map[string]api.Secret)

+}

@@ -252,12 +252,26 @@ func (s *session) watch(ctx context.Context) error {

 			}

 		}

 		if tasksWatch != nil {

+			// When falling back to Tasks because of an old managers, we wrap the tasks in assignments.

 			var taskResp *api.TasksMessage

+			var assignmentChanges []*api.AssignmentChange

 			taskResp, err = tasksWatch.Recv()

 			if err != nil {

 				return err

 			}

-			resp = &api.AssignmentsMessage{Type: api.AssignmentsMessage_COMPLETE, UpdateTasks: taskResp.Tasks}

+			for _, t := range taskResp.Tasks {

+				taskChange := &api.AssignmentChange{

+					Assignment: &api.Assignment{

+						Item: &api.Assignment_Task{

+							Task: t,

+						},

+					},

+					Action: api.AssignmentChange_AssignmentActionUpdate,

+				}

+

+				assignmentChanges = append(assignmentChanges, taskChange)

+			}

+			resp = &api.AssignmentsMessage{Type: api.AssignmentsMessage_COMPLETE, Changes: assignmentChanges}

 		}

 		// If there seems to be a gap in the stream, let's break out of the inner for and

@@ -17,13 +17,13 @@ type Worker interface {

 	// Init prepares the worker for task assignment.

 	Init(ctx context.Context) error

-	// AssignTasks assigns a complete set of tasks to a worker. Any task not included in

+	// Assign assigns a complete set of tasks and secrets to a worker. Any task or secrets not included in

 	// this set will be removed.

-	AssignTasks(ctx context.Context, tasks []*api.Task) error

+	Assign(ctx context.Context, assignments []*api.AssignmentChange) error

-	// UpdateTasks updates an incremental set of tasks to the worker. Any task not included

+	// Updates updates an incremental set of tasks or secrets of the worker. Any task/secret not included

 	// either in added or removed will remain untouched.

-	UpdateTasks(ctx context.Context, added []*api.Task, removed []string) error

+	Update(ctx context.Context, assignments []*api.AssignmentChange) error

 	// Listen to updates about tasks controlled by the worker. When first

 	// called, the reporter will receive all updates for all tasks controlled

@@ -42,6 +42,7 @@ type worker struct {

 	db        *bolt.DB

 	executor  exec.Executor

 	listeners map[*statusReporterKey]struct{}

+	secrets   *secrets

 	taskManagers map[string]*taskManager

 	mu           sync.RWMutex

@@ -53,6 +54,7 @@ func newWorker(db *bolt.DB, executor exec.Executor) *worker {

 		executor:     executor,

 		listeners:    make(map[*statusReporterKey]struct{}),

 		taskManagers: make(map[string]*taskManager),

+		secrets:      newSecrets(),

 	}

 }

@@ -90,37 +92,70 @@ func (w *worker) Init(ctx context.Context) error {

 	})

 }

-// AssignTasks assigns  the set of tasks to the worker. Any tasks not previously known will

-// be started. Any tasks that are in the task set and already running will be

-// updated, if possible. Any tasks currently running on the

-// worker outside the task set will be terminated.

-func (w *worker) AssignTasks(ctx context.Context, tasks []*api.Task) error {

+// Assign assigns a full set of tasks and secrets to the worker.

+// Any tasks not previously known will be started. Any tasks that are in the task set

+// and already running will be updated, if possible. Any tasks currently running on

+// the worker outside the task set will be terminated.

+// Any secrets not in the set of assignments will be removed.

+func (w *worker) Assign(ctx context.Context, assignments []*api.AssignmentChange) error {

 	w.mu.Lock()

 	defer w.mu.Unlock()

 	log.G(ctx).WithFields(logrus.Fields{

-		"len(tasks)": len(tasks),

-	}).Debug("(*worker).AssignTasks")

+		"len(assignments)": len(assignments),

+	}).Debug("(*worker).Assign")

-	return reconcileTaskState(ctx, w, tasks, nil, true)

+	// Need to update secrets before tasks, because tasks might depend on new secrets

+	err := reconcileSecrets(ctx, w, assignments, true)

+	if err != nil {

+		return err

+	}

+

+	return reconcileTaskState(ctx, w, assignments, true)

 }

-// UpdateTasks the set of tasks to the worker.

+// Update updates the set of tasks and secret for the worker.

 // Tasks in the added set will be added to the worker, and tasks in the removed set

 // will be removed from the worker

-func (w *worker) UpdateTasks(ctx context.Context, added []*api.Task, removed []string) error {

+// Serets in the added set will be added to the worker, and secrets in the removed set

+// will be removed from the worker.

+func (w *worker) Update(ctx context.Context, assignments []*api.AssignmentChange) error {

 	w.mu.Lock()

 	defer w.mu.Unlock()

 	log.G(ctx).WithFields(logrus.Fields{

-		"len(added)":   len(added),

-		"len(removed)": len(removed),

-	}).Debug("(*worker).UpdateTasks")

+		"len(assignments)": len(assignments),

+	}).Debug("(*worker).Update")

+

+	err := reconcileSecrets(ctx, w, assignments, false)

+	if err != nil {

+		return err

+	}

-	return reconcileTaskState(ctx, w, added, removed, false)

+	return reconcileTaskState(ctx, w, assignments, false)

 }

-func reconcileTaskState(ctx context.Context, w *worker, added []*api.Task, removed []string, fullSnapshot bool) error {

+func reconcileTaskState(ctx context.Context, w *worker, assignments []*api.AssignmentChange, fullSnapshot bool) error {

+	var (

+		updatedTasks []*api.Task

+		removedTasks []*api.Task

+	)

+	for _, a := range assignments {

+		if t := a.Assignment.GetTask(); t != nil {

+			switch a.Action {

+			case api.AssignmentChange_AssignmentActionUpdate:

+				updatedTasks = append(updatedTasks, t)

+			case api.AssignmentChange_AssignmentActionRemove:

+				removedTasks = append(removedTasks, t)

+			}

+		}

+	}

+

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

+		"len(updatedTasks)": len(updatedTasks),

+		"len(removedTasks)": len(removedTasks),

+	}).Debug("(*worker).reconcileTaskState")

+

 	tx, err := w.db.Begin(true)

 	if err != nil {

 		log.G(ctx).WithError(err).Error("failed starting transaction against task database")

@@ -130,7 +165,7 @@ func reconcileTaskState(ctx context.Context, w *worker, added []*api.Task, remov

 	assigned := map[string]struct{}{}

-	for _, task := range added {

+	for _, task := range updatedTasks {

 		log.G(ctx).WithFields(

 			logrus.Fields{

 				"task.id":           task.ID,

@@ -202,15 +237,15 @@ func reconcileTaskState(ctx context.Context, w *worker, added []*api.Task, remov

 	} else {

 		// If this was an incremental set of assignments, we're going to remove only the tasks

 		// in the removed set

-		for _, taskID := range removed {

-			err := removeTaskAssignment(taskID)

+		for _, task := range removedTasks {

+			err := removeTaskAssignment(task.ID)

 			if err != nil {

 				continue

 			}

-			tm, ok := w.taskManagers[taskID]

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

 			if ok {

-				delete(w.taskManagers, taskID)

+				delete(w.taskManagers, task.ID)

 				go closeManager(tm)

 			}

 		}

@@ -219,6 +254,39 @@ func reconcileTaskState(ctx context.Context, w *worker, added []*api.Task, remov

 	return tx.Commit()

 }

+func reconcileSecrets(ctx context.Context, w *worker, assignments []*api.AssignmentChange, fullSnapshot bool) error {

+	var (

+		updatedSecrets []api.Secret

+		removedSecrets []string

+	)

+	for _, a := range assignments {

+		if s := a.Assignment.GetSecret(); s != nil {

+			switch a.Action {

+			case api.AssignmentChange_AssignmentActionUpdate:

+				updatedSecrets = append(updatedSecrets, *s)

+			case api.AssignmentChange_AssignmentActionRemove:

+				removedSecrets = append(removedSecrets, s.ID)

+			}

+

+		}

+	}

+

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

+		"len(updatedSecrets)": len(updatedSecrets),

+		"len(removedSecrets)": len(removedSecrets),

+	}).Debug("(*worker).reconcileSecrets")

+

+	// If this was a complete set of secrets, we're going to clear the secrets map and add all of them

+	if fullSnapshot {

+		w.secrets.Reset()

+	} else {

+		w.secrets.Remove(removedSecrets)

+	}

+	w.secrets.Add(updatedSecrets...)

+

+	return nil

+}

+

 func (w *worker) Listen(ctx context.Context, reporter StatusReporter) {

 	w.mu.Lock()

 	defer w.mu.Unlock()

@@ -35,6 +35,29 @@ var _ = proto.Marshal

 var _ = fmt.Errorf

 var _ = math.Inf

+type AssignmentChange_AssignmentAction int32

+

+const (

+	AssignmentChange_AssignmentActionUpdate AssignmentChange_AssignmentAction = 0

+	AssignmentChange_AssignmentActionRemove AssignmentChange_AssignmentAction = 1

+)

+

+var AssignmentChange_AssignmentAction_name = map[int32]string{

+	0: "UPDATE",

+	1: "REMOVE",

+}

+var AssignmentChange_AssignmentAction_value = map[string]int32{

+	"UPDATE": 0,

+	"REMOVE": 1,

+}

+

+func (x AssignmentChange_AssignmentAction) String() string {

+	return proto.EnumName(AssignmentChange_AssignmentAction_name, int32(x))

+}

+func (AssignmentChange_AssignmentAction) EnumDescriptor() ([]byte, []int) {

+	return fileDescriptorDispatcher, []int{10, 0}

+}

+

 // AssignmentType specifies whether this assignment message carries

 // the full state, or is an update to an existing state.

 type AssignmentsMessage_Type int32

@@ -57,7 +80,7 @@ func (x AssignmentsMessage_Type) String() string {

 	return proto.EnumName(AssignmentsMessage_Type_name, int32(x))

 }

 func (AssignmentsMessage_Type) EnumDescriptor() ([]byte, []int) {

-	return fileDescriptorDispatcher, []int{9, 0}

+	return fileDescriptorDispatcher, []int{11, 0}

 }

 // SessionRequest starts a session.

@@ -214,6 +237,137 @@ func (m *AssignmentsRequest) Reset()                    { *m = AssignmentsReques

 func (*AssignmentsRequest) ProtoMessage()               {}

 func (*AssignmentsRequest) Descriptor() ([]byte, []int) { return fileDescriptorDispatcher, []int{8} }

+type Assignment struct {

+	// Types that are valid to be assigned to Item:

+	//	*Assignment_Task

+	//	*Assignment_Secret

+	Item isAssignment_Item `protobuf_oneof:"item"`

+}

+

+func (m *Assignment) Reset()                    { *m = Assignment{} }

+func (*Assignment) ProtoMessage()               {}

+func (*Assignment) Descriptor() ([]byte, []int) { return fileDescriptorDispatcher, []int{9} }

+

+type isAssignment_Item interface {

+	isAssignment_Item()

+	MarshalTo([]byte) (int, error)

+	Size() int

+}

+

+type Assignment_Task struct {

+	Task *Task `protobuf:"bytes,1,opt,name=task,oneof"`

+}

+type Assignment_Secret struct {

+	Secret *Secret `protobuf:"bytes,2,opt,name=secret,oneof"`

+}

+

+func (*Assignment_Task) isAssignment_Item()   {}

+func (*Assignment_Secret) isAssignment_Item() {}

+

+func (m *Assignment) GetItem() isAssignment_Item {

+	if m != nil {

+		return m.Item

+	}

+	return nil

+}

+

+func (m *Assignment) GetTask() *Task {

+	if x, ok := m.GetItem().(*Assignment_Task); ok {

+		return x.Task

+	}

+	return nil

+}

+

+func (m *Assignment) GetSecret() *Secret {

+	if x, ok := m.GetItem().(*Assignment_Secret); ok {

+		return x.Secret

+	}

+	return nil

+}

+

+// XXX_OneofFuncs is for the internal use of the proto package.

+func (*Assignment) XXX_OneofFuncs() (func(msg proto.Message, b *proto.Buffer) error, func(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error), func(msg proto.Message) (n int), []interface{}) {

+	return _Assignment_OneofMarshaler, _Assignment_OneofUnmarshaler, _Assignment_OneofSizer, []interface{}{

+		(*Assignment_Task)(nil),

+		(*Assignment_Secret)(nil),

+	}

+}

+

+func _Assignment_OneofMarshaler(msg proto.Message, b *proto.Buffer) error {

+	m := msg.(*Assignment)

+	// item

+	switch x := m.Item.(type) {

+	case *Assignment_Task:

+		_ = b.EncodeVarint(1<<3 | proto.WireBytes)

+		if err := b.EncodeMessage(x.Task); err != nil {

+			return err

+		}

+	case *Assignment_Secret:

+		_ = b.EncodeVarint(2<<3 | proto.WireBytes)

+		if err := b.EncodeMessage(x.Secret); err != nil {

+			return err

+		}

+	case nil:

+	default:

+		return fmt.Errorf("Assignment.Item has unexpected type %T", x)

+	}

+	return nil

+}

+

+func _Assignment_OneofUnmarshaler(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error) {

+	m := msg.(*Assignment)

+	switch tag {

+	case 1: // item.task

+		if wire != proto.WireBytes {

+			return true, proto.ErrInternalBadWireType

+		}

+		msg := new(Task)

+		err := b.DecodeMessage(msg)

+		m.Item = &Assignment_Task{msg}

+		return true, err

+	case 2: // item.secret

+		if wire != proto.WireBytes {

+			return true, proto.ErrInternalBadWireType

+		}

+		msg := new(Secret)

+		err := b.DecodeMessage(msg)

+		m.Item = &Assignment_Secret{msg}

+		return true, err

+	default:

+		return false, nil

+	}

+}

+

+func _Assignment_OneofSizer(msg proto.Message) (n int) {

+	m := msg.(*Assignment)

+	// item

+	switch x := m.Item.(type) {

+	case *Assignment_Task:

+		s := proto.Size(x.Task)

+		n += proto.SizeVarint(1<<3 | proto.WireBytes)

+		n += proto.SizeVarint(uint64(s))

+		n += s

+	case *Assignment_Secret:

+		s := proto.Size(x.Secret)

+		n += proto.SizeVarint(2<<3 | proto.WireBytes)

+		n += proto.SizeVarint(uint64(s))

+		n += s

+	case nil:

+	default:

+		panic(fmt.Sprintf("proto: unexpected type %T in oneof", x))

+	}

+	return n

+}

+

+type AssignmentChange struct {

+	Assignment *Assignment                       `protobuf:"bytes,1,opt,name=assignment" json:"assignment,omitempty"`

+	Action     AssignmentChange_AssignmentAction `protobuf:"varint,2,opt,name=action,proto3,enum=docker.swarmkit.v1.AssignmentChange_AssignmentAction" json:"action,omitempty"`

+}

+

+func (m *AssignmentChange) Reset()                    { *m = AssignmentChange{} }

+func (*AssignmentChange) ProtoMessage()               {}

+func (*AssignmentChange) Descriptor() ([]byte, []int) { return fileDescriptorDispatcher, []int{10} }

+

 type AssignmentsMessage struct {

 	Type AssignmentsMessage_Type `protobuf:"varint,1,opt,name=type,proto3,enum=docker.swarmkit.v1.AssignmentsMessage_Type" json:"type,omitempty"`

 	// AppliesTo references the previous ResultsIn value, to chain

@@ -226,28 +380,13 @@ type AssignmentsMessage struct {

 	// match against the next message's AppliesTo value and protect

 	// against missed messages.

 	ResultsIn string `protobuf:"bytes,3,opt,name=results_in,json=resultsIn,proto3" json:"results_in,omitempty"`

-	// UpdateTasks is a set of new or updated tasks to run on this node.

-	// In the first assignments message, it contains all of the tasks

-	// to run on this node. Tasks outside of this set running on the node

-	// should be terminated.

-	UpdateTasks []*Task `protobuf:"bytes,4,rep,name=update_tasks,json=updateTasks" json:"update_tasks,omitempty"`

-	// RemoveTasks is a set of previously-assigned task IDs to remove from the

-	// assignment set. It is not used in the first assignments message of

-	// a stream.

-	RemoveTasks []string `protobuf:"bytes,5,rep,name=remove_tasks,json=removeTasks" json:"remove_tasks,omitempty"`

-	// UpdateSecrets is a set of new or updated secrets for this node.

-	// In the first assignments message, it contains all of the secrets

-	// the node needs for itself and its assigned tasks.

-	UpdateSecrets []*Secret `protobuf:"bytes,6,rep,name=update_secrets,json=updateSecrets" json:"update_secrets,omitempty"`

-	// RemoveSecrets is a set of previously-assigned secret names to remove

-	// from memory. It is not used in the first assignments message of