core: update LastAllocUpdateIndex with allocs

Update allocs and the LastAllocUpdateIndex in the same Raft transaction
to avoid data inconsistency in case the UpdateAlloc request fails
midway.

Author: lgfa29

nomad/node_endpoint.go

@@ -1339,17 +1339,6 @@ func (n *Node) UpdateAlloc(args *structs.AllocUpdateRequest, reply *structs.Gene
 		return err
 	}
 
-	// Update node alloc update index.
-	copyNode := node.Copy()
-	copyNode.LastAllocUpdateIndex = future.Index()
-
-	_, _, err = n.srv.raftApply(structs.NodeRegisterRequestType, &structs.NodeRegisterRequest{
-		Node: copyNode,
-	})
-	if err != nil {
-		return fmt.Errorf("node update failed: %v", err)
-	}
-
 	// Setup the response
 	reply.Index = future.Index()
 	return nil

nomad/state/state_store.go

@@ -3602,8 +3602,13 @@ func (s *StateStore) UpdateAllocsFromClient(msgType structs.MessageType, index u
 	txn := s.db.WriteTxnMsgT(msgType, index)
 	defer txn.Abort()
 
+	// Capture all nodes being affected. Alloc updates from clients are batched
+	// so this request may include allocs from several nodes.
+	nodeIDs := make(map[string]interface{})
+
 	// Handle each of the updated allocations
 	for _, alloc := range allocs {
+		nodeIDs[alloc.NodeID] = struct{}{}
 		if err := s.nestedUpdateAllocFromClient(txn, index, alloc); err != nil {
 			return err
 		}
@@ -3614,6 +3619,13 @@ func (s *StateStore) UpdateAllocsFromClient(msgType structs.MessageType, index u
 		return fmt.Errorf("index update failed: %v", err)
 	}
 
+	// Update the index of when nodes last updated their allocs.
+	for nodeID := range nodeIDs {
+		if err := s.updateClientAllocUpdateIndex(txn, index, nodeID); err != nil {
+			return fmt.Errorf("node update failed: %v", err)
+		}
+	}
+
 	return txn.Commit()
 }
 
@@ -3705,6 +3717,28 @@ func (s *StateStore) nestedUpdateAllocFromClient(txn *txn, index uint64, alloc *
 	return nil
 }
 
+func (s *StateStore) updateClientAllocUpdateIndex(txn *txn, index uint64, nodeID string) error {
+	existing, err := txn.First("nodes", "id", nodeID)
+	if err != nil {
+		return fmt.Errorf("node lookup failed: %v", err)
+	}
+	if existing == nil {
+		return nil
+	}
+
+	node := existing.(*structs.Node)
+	copyNode := node.Copy()
+	copyNode.LastAllocUpdateIndex = index
+
+	if err := txn.Insert("nodes", copyNode); err != nil {
+		return fmt.Errorf("node update failed: %v", err)
+	}
+	if err := txn.Insert("index", &IndexEntry{"nodes", txn.Index}); err != nil {
+		return fmt.Errorf("index update failed: %v", err)
+	}
+	return nil
+}
+
 // UpsertAllocs is used to evict a set of allocations and allocate new ones at
 // the same time.
 func (s *StateStore) UpsertAllocs(msgType structs.MessageType, index uint64, allocs []*structs.Allocation) error {