Actually starting to use the deallocator to clean up services

**- What I did**

This patch actually stars to use the deallocator to clean up services, as
well as service-level resources (as of now, only networks).

**- How I did it**

A previous patch (moby#2759) introduced a
new component, the deallocator, responsible for cleaning up services and
service-level resources. This patch is actually starting to make use of that
component.

Since the deallocator used to rely on the reaper deleting the tasks belonging
to services that had been marked for removal, a previous version of this patch
was modifying the task reaper quite heavily to also keep track of such services
(needed since otherwise the reaper would fail to clean up all of them, instead
keeping some for history tracking purposes). However, it soon appeared that
this was not the best approach:
 * this creates a hidden coupling between the reaper and the deallocator
 * it's also not the best user experience to suddenly remove all of a service's
   task history while shutting down, for not apparent reason to the user

For these reasons, this patch instead amends the deallocator to also look at tasks status when keeping track of how many alive tasks remain to a service.

**- How to test it**

Updated tests.

**- Description for the changelog**

Services & networks are no longer deleted immediately when a user requests their
deletion; instead, they are deleted when all their tasks are actually shut down.

Signed-off-by: Jean Rouge <jer329@cornell.edu>

Author: wk8

cmd/swarmctl/network/inspect.go

@@ -51,6 +51,10 @@ func printNetworkSummary(network *api.Network) {
 	common.FprintfIfNotEmpty(w, "ID\t: %s\n", network.ID)
 	common.FprintfIfNotEmpty(w, "Name\t: %s\n", spec.Annotations.Name)
 
+	if network.PendingDelete {
+		common.FprintfIfNotEmpty(w, "[Network %s marked for removal]\n", spec.Annotations.Name)
+	}
+
 	fmt.Fprintln(w, "Spec:\t")
 	if len(spec.Annotations.Labels) > 0 {
 		fmt.Fprintln(w, "  Labels:\t")

cmd/swarmctl/service/inspect.go

@@ -22,12 +22,17 @@ func printServiceSummary(service *api.Service, running int) {
 	w := tabwriter.NewWriter(os.Stdout, 8, 8, 8, ' ', 0)
 	defer w.Flush()
 
-	task := service.Spec.Task
+	spec := service.Spec
 	common.FprintfIfNotEmpty(w, "ID\t: %s\n", service.ID)
-	common.FprintfIfNotEmpty(w, "Name\t: %s\n", service.Spec.Annotations.Name)
-	if len(service.Spec.Annotations.Labels) > 0 {
+	common.FprintfIfNotEmpty(w, "Name\t: %s\n", spec.Annotations.Name)
+
+	if service.PendingDelete {
+		common.FprintfIfNotEmpty(w, "[Service %s marked for removal]\n", spec.Annotations.Name)
+	}
+
+	if len(spec.Annotations.Labels) > 0 {
 		fmt.Fprintln(w, "Labels\t")
-		for k, v := range service.Spec.Annotations.Labels {
+		for k, v := range spec.Annotations.Labels {
 			fmt.Fprintf(w, "  %s\t: %s\n", k, v)
 		}
 	}
@@ -51,7 +56,8 @@ func printServiceSummary(service *api.Service, running int) {
 
 	fmt.Fprintln(w, "Template\t")
 	fmt.Fprintln(w, " Container\t")
-	ctr := service.Spec.Task.GetContainer()
+	task := spec.Task
+	ctr := task.GetContainer()
 	common.FprintfIfNotEmpty(w, "  Image\t: %s\n", ctr.Image)
 	common.FprintfIfNotEmpty(w, "  Command\t: %q\n", strings.Join(ctr.Command, " "))
 	common.FprintfIfNotEmpty(w, "  Args\t: [%s]\n", strings.Join(ctr.Args, ", "))
@@ -90,7 +96,7 @@ func printServiceSummary(service *api.Service, running int) {
 			printResources(w, res.Limits)
 		}
 	}
-	if len(service.Spec.Task.Networks) > 0 {
+	if len(spec.Task.Networks) > 0 {
 		fmt.Fprint(w, "  Networks:")
 		for _, n := range service.Spec.Task.Networks {
 			fmt.Fprintf(w, " %s", n.Target)

manager/controlapi/network.go

@@ -199,8 +199,10 @@ func (s *Server) removeNetwork(id string) error {
 			return status.Errorf(codes.Internal, "could not find services using network %s: %v", id, err)
 		}
 
-		if len(services) != 0 {
-			return status.Errorf(codes.FailedPrecondition, "network %s is in use by service %s", id, services[0].ID)
+		for _, service := range services {
+			if !service.PendingDelete {
+				return status.Errorf(codes.FailedPrecondition, "network %s is in use by service %s", id, service.ID)
+			}
 		}
 
 		tasks, err := store.FindTasks(tx, store.ByReferencedNetworkID(id))
@@ -214,7 +216,12 @@ func (s *Server) removeNetwork(id string) error {
 			}
 		}
 
-		return store.DeleteNetwork(tx, id)
+		network := store.GetNetwork(tx, id)
+		if network == nil {
+			return status.Errorf(codes.NotFound, "network %s not found", id)
+		}
+		network.PendingDelete = true
+		return store.UpdateNetwork(tx, network)
 	})
 }
 

manager/controlapi/network_test.go

@@ -191,11 +191,30 @@ func TestRemoveNetworkWithAttachedService(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotEqual(t, nr.Network, nil)
 	assert.NotEqual(t, nr.Network.ID, "")
-	createServiceInNetwork(t, ts, "name", "image", nr.Network.ID, 1)
+	createServiceInNetwork(t, ts, "service1", "image", nr.Network.ID, 1)
 	_, err = ts.Client.RemoveNetwork(context.Background(), &api.RemoveNetworkRequest{NetworkID: nr.Network.ID})
 	assert.Error(t, err)
 }
 
+func TestRemoveNetworkWithAttachedServiceMarkedForRemoval(t *testing.T) {
+	ts := newTestServer(t)
+	defer ts.Stop()
+	nr, err := ts.Client.CreateNetwork(context.Background(), &api.CreateNetworkRequest{
+		Spec: createNetworkSpec("testnet5"),
+	})
+	assert.NoError(t, err)
+	assert.NotEqual(t, nr.Network, nil)
+	assert.NotEqual(t, nr.Network.ID, "")
+	service := createServiceInNetwork(t, ts, "service2", "image", nr.Network.ID, 1)
+	// then let's delete the service
+	r, err := ts.Client.RemoveService(context.Background(), &api.RemoveServiceRequest{ServiceID: service.ID})
+	assert.NoError(t, err)
+	assert.NotNil(t, r)
+	// now we should be able to delete the network
+	_, err = ts.Client.RemoveNetwork(context.Background(), &api.RemoveNetworkRequest{NetworkID: nr.Network.ID})
+	assert.NoError(t, err)
+}
+
 func TestCreateNetworkInvalidDriver(t *testing.T) {
 	ts := newTestServer(t)
 	defer ts.Stop()

manager/controlapi/service.go

@@ -718,6 +718,7 @@ func (s *Server) GetService(ctx context.Context, request *api.GetServiceRequest)
 // - Returns `NotFound` if the Service is not found.
 // - Returns `InvalidArgument` if the ServiceSpec is malformed.
 // - Returns `Unimplemented` if the ServiceSpec references unimplemented features.
+// - Returns `FailedPrecondition` if the Service is marked for removal
 // - Returns an error if the update fails.
 func (s *Server) UpdateService(ctx context.Context, request *api.UpdateServiceRequest) (*api.UpdateServiceResponse, error) {
 	if request.ServiceID == "" || request.ServiceVersion == nil {
@@ -751,6 +752,10 @@ func (s *Server) UpdateService(ctx context.Context, request *api.UpdateServiceRe
 			return status.Errorf(codes.NotFound, "service %s not found", request.ServiceID)
 		}
 
+		if service.PendingDelete {
+			return status.Errorf(codes.FailedPrecondition, "service %s is marked for removal", request.ServiceID)
+		}
+
 		// It's not okay to update Service.Spec.Networks on its own.
 		// However, if Service.Spec.Task.Networks is also being
 		// updated, that's okay (for example when migrating from the
@@ -844,12 +849,15 @@ func (s *Server) RemoveService(ctx context.Context, request *api.RemoveServiceRe
 	}
 
 	err := s.store.Update(func(tx store.Tx) error {
-		return store.DeleteService(tx, request.ServiceID)
+		service := store.GetService(tx, request.ServiceID)
+		if service == nil {
+			return status.Errorf(codes.NotFound, "service %s not found", request.ServiceID)
+		}
+		// mark service for removal
+		service.PendingDelete = true
+		return store.UpdateService(tx, service)
 	})
 	if err != nil {
-		if err == store.ErrNotExist {
-			return nil, status.Errorf(codes.NotFound, "service %s not found", request.ServiceID)
-		}
 		return nil, err
 	}
 	return &api.RemoveServiceResponse{}, nil

manager/controlapi/service_test.go

@@ -916,6 +916,19 @@ func TestUpdateService(t *testing.T) {
 	assert.Equal(t, codes.InvalidArgument, testutils.ErrorCode(err))
 }
 
+func TestUpdateServiceMarkedForRemoval(t *testing.T) {
+	ts := newTestServer(t)
+	defer ts.Stop()
+
+	service := createService(t, ts, "name", "image", 1)
+	r, err := ts.Client.RemoveService(context.Background(), &api.RemoveServiceRequest{ServiceID: service.ID})
+	assert.NoError(t, err)
+	assert.NotNil(t, r)
+
+	_, err = ts.Client.UpdateService(context.Background(), &api.UpdateServiceRequest{ServiceID: service.ID, Spec: &service.Spec, ServiceVersion: &service.Meta.Version})
+	assert.Equal(t, codes.FailedPrecondition, testutils.ErrorCode(err))
+}
+
 func TestServiceUpdateRejectNetworkChange(t *testing.T) {
 	ts := newTestServer(t)
 	defer ts.Stop()

manager/deallocator/deallocator.go

@@ -2,6 +2,7 @@ package deallocator
 
 import (
 	"context"
+	"sync"
 
 	"github.com/docker/go-events"
 	"github.com/docker/swarmkit/api"
@@ -32,6 +33,9 @@ import (
 type Deallocator struct {
 	store *store.MemoryStore
 
+	// closeOnce ensures that stopChan is closed only once
+	closeOnce sync.Once
+
 	// for services that are shutting down, we keep track of how many
 	// tasks still exist for them
 	services map[string]*serviceWithTaskCounts
@@ -94,6 +98,7 @@ func (deallocator *Deallocator) Run(ctx context.Context) error {
 
 			return
 		},
+		api.EventUpdateTask{},
 		api.EventDeleteTask{},
 		api.EventUpdateService{},
 		api.EventUpdateNetwork{})
@@ -129,7 +134,7 @@ func (deallocator *Deallocator) Run(ctx context.Context) error {
 	for {
 		select {
 		case event := <-eventsChan:
-			if updated, err := deallocator.processNewEvent(ctx, event); err == nil {
+			if updated, err := deallocator.handleEvent(ctx, event); err == nil {
 				deallocator.notifyEventChan(updated)
 			} else {
 				log.G(ctx).WithError(err).Errorf("error processing deallocator event %#v", event)
@@ -142,11 +147,16 @@ func (deallocator *Deallocator) Run(ctx context.Context) error {
 	}
 }
 
-// Stop stops the deallocator's routine
-// FIXME (jrouge): see the comment on TaskReaper.Stop() and see when to properly stop this
-// plus unit test on this!
+// Stop stops the deallocator's routine and wait for the main loop to exit
+// Stop can be called in two cases. One when the manager is
+// shutting down, and the other when the manager (the leader) is
+// becoming a follower. Since these two instances could race with
+// each other, we use closeOnce here to ensure that TaskReaper.Stop()
+// is called only once to avoid a panic.
 func (deallocator *Deallocator) Stop() {
-	close(deallocator.stopChan)
+	deallocator.closeOnce.Do(func() {
+		close(deallocator.stopChan)
+	})
 	<-deallocator.doneChan
 }
 
@@ -180,12 +190,21 @@ func (deallocator *Deallocator) processService(ctx context.Context, service *api
 		// better to clean up resources that shouldn't be cleaned up yet
 		// than ending up with a service and some resources lost in limbo forever
 		return true, deallocator.deallocateService(ctx, service)
-	} else if len(tasks) == 0 {
-		// no tasks remaining for this service, we can clean it up
-		return true, deallocator.deallocateService(ctx, service)
+	} else {
+		remainingTasks := 0
+		for _, task := range tasks {
+			if isTaskStillAlive(task) {
+				remainingTasks++
+			}
+		}
+
+		if remainingTasks == 0 {
+			// no tasks remaining for this service, we can clean it up
+			return true, deallocator.deallocateService(ctx, service)
+		}
+		deallocator.services[service.ID] = &serviceWithTaskCounts{service: service, taskCount: remainingTasks}
+		return false, nil
 	}
-	deallocator.services[service.ID] = &serviceWithTaskCounts{service: service, taskCount: len(tasks)}
-	return false, nil
 }
 
 func (deallocator *Deallocator) deallocateService(ctx context.Context, service *api.Service) (err error) {
@@ -263,24 +282,16 @@ func (deallocator *Deallocator) processNetwork(ctx context.Context, tx store.Tx,
 	return
 }
 
-// Processes new events, and dispatches to the right method depending on what
+// Handles new events, and dispatches to the right method depending on what
 // type of event it is.
 // The boolean part of the return tuple indicates whether anything was actually
 // removed from the store
-func (deallocator *Deallocator) processNewEvent(ctx context.Context, event events.Event) (bool, error) {
+func (deallocator *Deallocator) handleEvent(ctx context.Context, event events.Event) (bool, error) {
 	switch typedEvent := event.(type) {
+	case api.EventUpdateTask:
+		return deallocator.processTaskEvent(ctx, typedEvent.Task, typedEvent.OldTask)
 	case api.EventDeleteTask:
-		serviceID := typedEvent.Task.ServiceID
-
-		if serviceWithCount, present := deallocator.services[serviceID]; present {
-			if serviceWithCount.taskCount <= 1 {
-				delete(deallocator.services, serviceID)
-				return deallocator.processService(ctx, serviceWithCount.service)
-			}
-			serviceWithCount.taskCount--
-		}
-
-		return false, nil
+		return deallocator.processTaskEvent(ctx, nil, typedEvent.Task)
 	case api.EventUpdateService:
 		return deallocator.processService(ctx, typedEvent.Service)
 	case api.EventUpdateNetwork:
@@ -289,3 +300,30 @@ func (deallocator *Deallocator) processNewEvent(ctx context.Context, event event
 		return false, nil
 	}
 }
+
+// Common logic for handling task update/delete events
+// newTask is nil for delete events
+func (deallocator *Deallocator) processTaskEvent(ctx context.Context, newTask, oldTask *api.Task) (bool, error) {
+	serviceID := oldTask.ServiceID
+	serviceWithCount, present := deallocator.services[serviceID]
+
+	if present && isTaskStillAlive(oldTask) && (newTask == nil || !isTaskStillAlive(newTask)) {
+		// this task belongs to a service that's shutting down, and was amongst
+		// the ones that we counted as alive when adding this service to our
+		// internal state
+
+		if serviceWithCount.taskCount <= 1 {
+			delete(deallocator.services, serviceID)
+			return deallocator.processService(ctx, serviceWithCount.service)
+		}
+		serviceWithCount.taskCount--
+	}
+
+	return false, nil
+}
+
+// simple helper function to distinguish tasks that are still running
+// from ones that are done
+func isTaskStillAlive(task *api.Task) bool {
+	return task.Status.State <= api.TaskStateRunning
+}