[18.03 backport] Fix leaking task resources when nodes are deleted

manager/controlapi/node.go

@@ -3,6 +3,7 @@ package controlapi
 import (
 	"crypto/x509"
 	"encoding/pem"
+	"time"
 
 	"github.com/docker/swarmkit/api"
 	"github.com/docker/swarmkit/manager/state/raft/membership"
@@ -248,24 +249,45 @@ func (s *Server) UpdateNode(ctx context.Context, request *api.UpdateNodeRequest)
 	}, nil
 }
 
-func removeNodeAttachments(tx store.Tx, nodeID string) error {
-	// orphan the node's attached containers. if we don't do this, the
-	// network these attachments are connected to will never be removeable
+func orphanNodeTasks(tx store.Tx, nodeID string) error {
+	// when a node is deleted, all of its tasks are irrecoverably removed.
+	// additionally, the Dispatcher can no longer be relied on to update the
+	// task status. Therefore, when the node is removed, we must additionally
+	// move all of its assigned tasks to the Orphaned state, so that their
+	// resources can be cleaned up.
 	tasks, err := store.FindTasks(tx, store.ByNodeID(nodeID))
 	if err != nil {
 		return err
 	}
 	for _, task := range tasks {
-		// if the task is an attachment, then we just delete it. the allocator
-		// will do the heavy lifting. basically, GetAttachment will return the
-		// attachment if that's the kind of runtime, or nil if it's not.
-		if task.Spec.GetAttachment() != nil {
-			// don't delete the task. instead, update it to `ORPHANED` so that
-			// the taskreaper will clean it up.
-			task.Status.State = api.TaskStateOrphaned
-			if err := store.UpdateTask(tx, task); err != nil {
-				return err
+		// this operation must occur within the same transaction boundary. If
+		// we cannot accomplish this task orphaning in the same transaction, we
+		// could crash or die between transactions and not get a chance to do
+		// this. however, in cases were there is an exceptionally large number
+		// of tasks for a node, this may cause the transaction to exceed the
+		// max message size.
+		//
+		// therefore, we restrict updating to only tasks in a non-terminal
+		// state. Tasks in a terminal state do not need to be updated.
+		if task.Status.State < api.TaskStateCompleted {
+			// this code was backported from a later version. in the later version,
+			// gogotypes has a function TimestampNow, which returns a
+			// gogotypes.Timestamp for the current time. this version of the
+			// swarmkit code uses an earlier version of gogotypes, which means we
+			// don't have access to that function. however, this code is
+			// esssentially equivalent.
+			//
+			// we're ignoring the error value of TimestampProto, because it's
+			// probably a catastrophic error if time.Now doesn't return a valid
+			// time, and gogotypes.TimestampNow just panics if the conversion
+			// fails anyway.
+			timestamp, _ := gogotypes.TimestampProto(time.Now())
+			task.Status = api.TaskStatus{
+				Timestamp: timestamp,
+				State:     api.TaskStateOrphaned,
+				Message:   "Task belonged to a node that has been deleted",
 			}
+			store.UpdateTask(tx, task)
 		}
 	}
 	return nil
@@ -336,7 +358,7 @@ func (s *Server) RemoveNode(ctx context.Context, request *api.RemoveNodeRequest)
 			return err
 		}
 
-		if err := removeNodeAttachments(tx, request.NodeID); err != nil {
+		if err := orphanNodeTasks(tx, request.NodeID); err != nil {
 			return err
 		}
 

manager/controlapi/node_test.go

@@ -732,10 +732,8 @@ func TestUpdateNodeDemote(t *testing.T) {
 	testUpdateNodeDemote(t)
 }
 
-// TestRemoveNodeAttachments tests the unexported removeNodeAttachments
-// function. This avoids us having to update the TestRemoveNodes function to
-// test all of this logic
-func TestRemoveNodeAttachments(t *testing.T) {
+// TestRemoveNodeAttachments tests the unexported orphanNodeTasks
+func TestOrphanNodeTasks(t *testing.T) {
 	// first, set up a store and all that
 	ts := newTestServer(t)
 	defer ts.Stop()
@@ -873,28 +871,50 @@ func TestRemoveNodeAttachments(t *testing.T) {
 				},
 			},
 		}
-		return store.CreateTask(tx, task4)
+		if err := store.CreateTask(tx, task4); err != nil {
+			return err
+		}
+
+		// 5.) A regular task that's already in a terminal state on the node,
+		//	   which does not need to be updated.
+		task5 := &api.Task{
+			ID:           "task5",
+			NodeID:       "id2",
+			DesiredState: api.TaskStateRunning,
+			Status: api.TaskStatus{
+				// use TaskStateCompleted, as this is the earliest terminal
+				// state (this ensures we don't actually use <= instead of <)
+				State: api.TaskStateCompleted,
+			},
+			Spec: api.TaskSpec{
+				Runtime: &api.TaskSpec_Container{
+					Container: &api.ContainerSpec{},
+				},
+			},
+		}
+		return store.CreateTask(tx, task5)
 	})
 	require.NoError(t, err)
 
 	// Now, call the function with our nodeID. make sure it returns no error
 	err = ts.Store.Update(func(tx store.Tx) error {
-		return removeNodeAttachments(tx, "id2")
+		return orphanNodeTasks(tx, "id2")
 	})
 	require.NoError(t, err)
 
-	// Now, make sure only task1, the network-attacahed task on id2, was
-	// removed
+	// Now, make sure only tasks 1 and 3, the tasks on the node we're deleting
+	// removed, are removed
 	ts.Store.View(func(tx store.ReadTx) {
 		tasks, err := store.FindTasks(tx, store.All)
 		require.NoError(t, err)
-		// should only be 3 tasks left
-		require.Len(t, tasks, 4)
-		// and the list should not contain task1
+		require.Len(t, tasks, 5)
+		// and the list should not contain task1 or task2
 		for _, task := range tasks {
 			require.NotNil(t, task)
-			if task.ID == "task1" {
+			if task.ID == "task1" || task.ID == "task3" {
 				require.Equal(t, task.Status.State, api.TaskStateOrphaned)
+			} else {
+				require.NotEqual(t, task.Status.State, api.TaskStateOrphaned)
 			}
 		}
 	})