orchestrator: Use task history to evaluate restart policy

api/types.proto

@@ -514,6 +514,15 @@ message TaskStatus {
 	// HostPorts provides a list of ports allocated at the host
 	// level.
 	PortStatus port_status = 6;
+
+	// AppliedBy gives the node ID of the manager that applied this task
+	// status update to the Task object.
+	string applied_by = 7;
+
+	// AppliedAt gives a timestamp of when this status update was applied to
+	// the Task object.
+	// Note: can't use stdtime because this field is nullable.
+	google.protobuf.Timestamp applied_at = 8;
 }
 
 // NetworkAttachmentConfig specifies how a service should be attached to a particular network.

manager/dispatcher/dispatcher.go

@@ -19,6 +19,7 @@ import (
 	"github.com/docker/swarmkit/log"
 	"github.com/docker/swarmkit/manager/drivers"
 	"github.com/docker/swarmkit/manager/state/store"
+	"github.com/docker/swarmkit/protobuf/ptypes"
 	"github.com/docker/swarmkit/remotes"
 	"github.com/docker/swarmkit/watch"
 	gogotypes "github.com/gogo/protobuf/types"
@@ -127,6 +128,7 @@ type Dispatcher struct {
 	cancel               context.CancelFunc
 	clusterUpdateQueue   *watch.Queue
 	dp                   *drivers.DriverProvider
+	securityConfig       *ca.SecurityConfig
 
 	taskUpdates     map[string]*api.TaskStatus // indexed by task ID
 	taskUpdatesLock sync.Mutex
@@ -144,7 +146,7 @@ type Dispatcher struct {
 }
 
 // New returns Dispatcher with cluster interface(usually raft.Node).
-func New(cluster Cluster, c *Config, dp *drivers.DriverProvider) *Dispatcher {
+func New(cluster Cluster, c *Config, dp *drivers.DriverProvider, securityConfig *ca.SecurityConfig) *Dispatcher {
 	d := &Dispatcher{
 		dp:                    dp,
 		nodes:                 newNodeStore(c.HeartbeatPeriod, c.HeartbeatEpsilon, c.GracePeriodMultiplier, c.RateLimitPeriod),
@@ -153,6 +155,7 @@ func New(cluster Cluster, c *Config, dp *drivers.DriverProvider) *Dispatcher {
 		cluster:               cluster,
 		processUpdatesTrigger: make(chan struct{}, 1),
 		config:                c,
+		securityConfig:        securityConfig,
 	}
 
 	d.processUpdatesCond = sync.NewCond(&d.processUpdatesLock)
@@ -630,6 +633,8 @@ func (d *Dispatcher) processUpdates(ctx context.Context) {
 				}
 
 				task.Status = *status
+				task.Status.AppliedBy = d.securityConfig.ClientTLSCreds.NodeID()
+				task.Status.AppliedAt = ptypes.MustTimestampProto(time.Now())
 				if err := store.UpdateTask(tx, task); err != nil {
 					logger.WithError(err).Error("failed to update task status")
 					return nil

manager/dispatcher/dispatcher_test.go

@@ -155,7 +155,7 @@ func startDispatcher(c *Config) (*grpcDispatcher, error) {
 	s := grpc.NewServer(serverOpts...)
 	tc := newTestCluster(l.Addr().String(), tca.MemoryStore)
 	driverGetter := &mockPluginGetter{}
-	d := New(tc, c, drivers.New(driverGetter))
+	d := New(tc, c, drivers.New(driverGetter), managerSecurityConfig)
 
 	authorize := func(ctx context.Context, roles []string) error {
 		_, err := ca.AuthorizeForwardedRoleAndOrg(ctx, roles, []string{ca.ManagerRole}, tca.Organization, nil)

manager/manager.go

@@ -225,7 +225,7 @@ func New(config *Config) (*Manager, error) {
 	m := &Manager{
 		config:          *config,
 		caserver:        ca.NewServer(raftNode.MemoryStore(), config.SecurityConfig, config.RootCAPaths),
-		dispatcher:      dispatcher.New(raftNode, dispatcher.DefaultConfig(), drivers.New(config.PluginGetter)),
+		dispatcher:      dispatcher.New(raftNode, dispatcher.DefaultConfig(), drivers.New(config.PluginGetter), config.SecurityConfig),
 		logbroker:       logbroker.New(raftNode.MemoryStore()),
 		watchServer:     watchapi.NewServer(raftNode.MemoryStore()),
 		server:          grpc.NewServer(opts...),

manager/orchestrator/global/global.go

@@ -255,6 +255,11 @@ func (g *Orchestrator) reconcileServices(ctx context.Context, serviceIDs []strin
 
 	g.store.View(func(tx store.ReadTx) {
 		for _, serviceID := range serviceIDs {
+			service := g.globalServices[serviceID].Service
+			if service == nil {
+				continue
+			}
+
 			tasks, err := store.FindTasks(tx, store.ByServiceID(serviceID))
 			if err != nil {
 				log.G(ctx).WithError(err).Errorf("global orchestrator: reconcileServices failed finding tasks for service %s", serviceID)
@@ -265,11 +270,22 @@ func (g *Orchestrator) reconcileServices(ctx context.Context, serviceIDs []strin
 			nodeTasks[serviceID] = make(map[string][]*api.Task)
 
 			for _, t := range tasks {
-				if t.DesiredState <= api.TaskStateRunning {
-					// Collect all running instances of this service
-					nodeTasks[serviceID][t.NodeID] = append(nodeTasks[serviceID][t.NodeID], t)
+				nodeTasks[serviceID][t.NodeID] = append(nodeTasks[serviceID][t.NodeID], t)
+			}
+
+			// Keep all runnable instances of this service,
+			// and instances that were not be restarted due
+			// to restart policy but may be updated if the
+			// service spec changed.
+			for nodeID, slot := range nodeTasks[serviceID] {
+				updatable := g.restarts.UpdatableTasksInSlot(ctx, slot, g.globalServices[serviceID].Service)
+				if len(updatable) != 0 {
+					nodeTasks[serviceID][nodeID] = updatable
+				} else {
+					delete(nodeTasks[serviceID], nodeID)
 				}
 			}
+
 		}
 	})
 
@@ -374,11 +390,6 @@ func (g *Orchestrator) reconcileOneNode(ctx context.Context, node *api.Node) {
 		return
 	}
 
-	var serviceIDs []string
-	for id := range g.globalServices {
-		serviceIDs = append(serviceIDs, id)
-	}
-
 	node, exists := g.nodes[node.ID]
 	if !exists {
 		return
@@ -400,24 +411,31 @@ func (g *Orchestrator) reconcileOneNode(ctx context.Context, node *api.Node) {
 		return
 	}
 
-	for _, serviceID := range serviceIDs {
+	for serviceID, service := range g.globalServices {
 		for _, t := range tasksOnNode {
 			if t.ServiceID != serviceID {
 				continue
 			}
-			if t.DesiredState <= api.TaskStateRunning {
-				tasks[serviceID] = append(tasks[serviceID], t)
+			tasks[serviceID] = append(tasks[serviceID], t)
+		}
+
+		// Keep all runnable instances of this service,
+		// and instances that were not be restarted due
+		// to restart policy but may be updated if the
+		// service spec changed.
+		for serviceID, slot := range tasks {
+			updatable := g.restarts.UpdatableTasksInSlot(ctx, slot, service.Service)
+
+			if len(updatable) != 0 {
+				tasks[serviceID] = updatable
+			} else {
+				delete(tasks, serviceID)
 			}
 		}
 	}
 
 	err = g.store.Batch(func(batch *store.Batch) error {
-		for _, serviceID := range serviceIDs {
-			service, exists := g.globalServices[serviceID]
-			if !exists {
-				continue
-			}
-
+		for serviceID, service := range g.globalServices {
 			if !constraint.NodeMatches(service.constraints, node) {
 				continue
 			}
@@ -560,6 +578,14 @@ func (g *Orchestrator) IsRelatedService(service *api.Service) bool {
 	return orchestrator.IsGlobalService(service)
 }
 
+// SlotTuple returns a slot tuple for the global service task.
+func (g *Orchestrator) SlotTuple(t *api.Task) orchestrator.SlotTuple {
+	return orchestrator.SlotTuple{
+		ServiceID: t.ServiceID,
+		NodeID:    t.NodeID,
+	}
+}
+
 func isTaskCompleted(t *api.Task, restartPolicy api.RestartPolicy_RestartCondition) bool {
 	if t == nil || t.DesiredState <= api.TaskStateRunning {
 		return false