Serialize NIC add/remove on the target VM's work job queue

addVmToNetwork() runs directly (no queue) whenever the calling thread is
inside any VM work job:

    jobContext.isJobDispatchedBy(VM_WORK_JOB_DISPATCHER)

That check does not care whose work job it is. When several VMs deploy in
parallel into different tiers of a VPC, each VmWorkStart job implements its
tier and calls addVpcRouterToGuestNetwork() -> addVmToNetwork(router, ...).
The check is true (we are in the user VM's work job), so every tier attach
runs orchestrateAddVmToNetwork() on the router directly. Nothing serializes
them:

- concurrent getFreeDeviceId() calls hand the same device id to several NICs
- the colliding PlugNicCommands partially fail, then cleanup and retries
  reshuffle device ids differently on each router
- on a redundant VPC the two routers' keepalived configs diverge and VRRP
  breaks: FAULT or both PRIMARY

Fix: run directly only when the current work job belongs to the VM being
modified (new helper isRunningVmWorkJobForVm()); otherwise dispatch through
that VM's own work job queue. Apply the same rule to removeNicFromVm() and
removeVmFromNetwork() — the latter always ran directly. Scope
removeVmFromNetworkThroughJobQueue()'s pending-job lookup by network uuid,
like the add path, so removals of different networks do not collapse into
one job. Drop an unused retrievePendingWorkJob() call from
addVmToNetworkThroughJobQueue().

Fixes: #11710

Signed-off-by: Brad House <bhouse@nexthop.ai>

Author: bhouse-nexthop

engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java

@@ -4549,12 +4549,29 @@ public boolean upgradeVmDb(final long vmId, final ServiceOffering newServiceOffe
         return _vmDao.update(vmId, vmForUpdate);
     }
 
+    /**
+     * A VM operation may run directly only when the current thread is already
+     * executing the work job of that same VM, as that VM's work queue is then
+     * held and provides the serialization. Running inside some other VM's work
+     * job (for example a router NIC operation performed while a user VM's
+     * start job implements a VPC tier) provides no such guarantee, so the
+     * operation must be dispatched through the target VM's own work job queue
+     * instead.
+     */
+    protected boolean isRunningVmWorkJobForVm(final AsyncJobExecutionContext jobContext, final long vmId) {
+        if (!jobContext.isJobDispatchedBy(VmWorkConstants.VM_WORK_JOB_DISPATCHER)) {
+            return false;
+        }
+        final VmWorkJobVO workJob = _workJobDao.findById(jobContext.getJob().getId());
+        return workJob != null && workJob.getVmInstanceId() == vmId;
+    }
+
     @Override
     public NicProfile addVmToNetwork(final VirtualMachine vm, final Network network, final NicProfile requested)
             throws ConcurrentOperationException, ResourceUnavailableException, InsufficientCapacityException {
 
         final AsyncJobExecutionContext jobContext = AsyncJobExecutionContext.getCurrentExecutionContext();
-        if (jobContext.isJobDispatchedBy(VmWorkConstants.VM_WORK_JOB_DISPATCHER)) {
+        if (isRunningVmWorkJobForVm(jobContext, vm.getId())) {
             VmWorkJobVO placeHolder = createPlaceHolderWork(vm.getId(), network.getUuid());
             try {
                 return orchestrateAddVmToNetwork(vm, network, requested);
@@ -4661,7 +4678,7 @@ public boolean removeNicFromVm(final VirtualMachine vm, final Nic nic)
             throws ConcurrentOperationException, ResourceUnavailableException {
 
         final AsyncJobExecutionContext jobContext = AsyncJobExecutionContext.getCurrentExecutionContext();
-        if (jobContext.isJobDispatchedBy(VmWorkConstants.VM_WORK_JOB_DISPATCHER)) {
+        if (isRunningVmWorkJobForVm(jobContext, vm.getId())) {
             VmWorkJobVO placeHolder = createPlaceHolderWork(vm.getId());
             try {
                 return orchestrateRemoveNicFromVm(vm, nic);
@@ -4737,7 +4754,25 @@ private boolean orchestrateRemoveNicFromVm(final VirtualMachine vm, final Nic ni
     @Override
     @DB
     public boolean removeVmFromNetwork(final VirtualMachine vm, final Network network, final URI broadcastUri) throws ConcurrentOperationException, ResourceUnavailableException {
-        return orchestrateRemoveVmFromNetwork(vm, network, broadcastUri);
+        final AsyncJobExecutionContext jobContext = AsyncJobExecutionContext.getCurrentExecutionContext();
+        if (isRunningVmWorkJobForVm(jobContext, vm.getId())) {
+            return orchestrateRemoveVmFromNetwork(vm, network, broadcastUri);
+        }
+
+        final Outcome<VirtualMachine> outcome = removeVmFromNetworkThroughJobQueue(vm, network, broadcastUri);
+
+        retrieveVmFromJobOutcome(outcome, vm.getUuid(), "removeVmFromNetwork");
+
+        try {
+            Object jobResult = retrieveResultFromJobOutcomeAndThrowExceptionIfNeeded(outcome);
+            if (jobResult != null && jobResult instanceof Boolean) {
+                return (Boolean)jobResult;
+            }
+        } catch (InsufficientCapacityException ex) {
+            throw new RuntimeException("Unexpected exception", ex);
+        }
+
+        throw new RuntimeException("Job failed with un-handled exception");
     }
 
     @DB
@@ -5887,10 +5922,6 @@ public Outcome<VirtualMachine> migrateVmStorageThroughJobQueue(final String vmUu
 
     public Outcome<VirtualMachine> addVmToNetworkThroughJobQueue(
             final VirtualMachine vm, final Network network, final NicProfile requested) {
-        Long vmId = vm.getId();
-        String commandName = VmWorkAddVmToNetwork.class.getName();
-        Pair<VmWorkJobVO, Long> pendingWorkJob = retrievePendingWorkJob(vmId, commandName);
-
         final CallContext context = CallContext.current();
         final User user = context.getCallingUser();
         final Account account = context.getCallingAccount();
@@ -5989,15 +6020,21 @@ public Outcome<VirtualMachine> removeVmFromNetworkThroughJobQueue(
             final VirtualMachine vm, final Network network, final URI broadcastUri) {
         Long vmId = vm.getId();
         String commandName = VmWorkRemoveVmFromNetwork.class.getName();
-        Pair<VmWorkJobVO, Long> pendingWorkJob = retrievePendingWorkJob(vmId, commandName);
 
-        VmWorkJobVO workJob = pendingWorkJob.first();
+        // pending-job lookup must be scoped by network, otherwise concurrent
+        // removals for different networks on the same VM collapse into one job
+        final List<VmWorkJobVO> pendingWorkJobs = _workJobDao.listPendingWorkJobs(
+                VirtualMachine.Type.Instance, vmId, commandName, network.getUuid());
 
-        if (workJob == null) {
+        VmWorkJobVO workJob;
+        if (CollectionUtils.isNotEmpty(pendingWorkJobs)) {
+            workJob = pendingWorkJobs.get(0);
+        } else {
             Pair<VmWorkJobVO, VmWork> newVmWorkJobAndInfo = createWorkJobAndWorkInfo(commandName, vmId);
 
             workJob = newVmWorkJobAndInfo.first();
-            VmWorkRemoveVmFromNetwork workInfo = new VmWorkRemoveVmFromNetwork(newVmWorkJobAndInfo.second(), network, broadcastUri);
+            workJob.setSecondaryObjectIdentifier(network.getUuid());
+            VmWorkRemoveVmFromNetwork workInfo = new VmWorkRemoveVmFromNetwork(newVmWorkJobAndInfo.second(), network.getId(), broadcastUri);
 
             setCmdInfoAndSubmitAsyncJob(workJob, workInfo, vmId);
         }
@@ -6132,7 +6169,7 @@ private Pair<JobInfo.Status, String> orchestrateRemoveNicFromVm(final VmWorkRemo
     private Pair<JobInfo.Status, String> orchestrateRemoveVmFromNetwork(final VmWorkRemoveVmFromNetwork work) throws Exception {
         VMInstanceVO vm = findVmById(work.getVmId());
         final boolean result = orchestrateRemoveVmFromNetwork(vm,
-                work.getNetwork(), work.getBroadcastUri());
+                _networkDao.findById(work.getNetworkId()), work.getBroadcastUri());
         return new Pair<>(JobInfo.Status.SUCCEEDED,
                 _jobMgr.marshallResultObject(result));
     }

engine/orchestration/src/main/java/com/cloud/vm/VmWorkRemoveVmFromNetwork.java

@@ -18,32 +18,33 @@
 
 import java.net.URI;
 
-import com.cloud.network.Network;
-
 public class VmWorkRemoveVmFromNetwork extends VmWork {
     private static final long serialVersionUID = -5070392905642149925L;
 
-    Network network;
-    URI broadcastUri;
+    // only the network id and the broadcast uri as text are stored: the work
+    // info is serialized with Gson, and entity or URI object graphs do not
+    // survive that (see VmWorkAddVmToNetwork, which stores the id the same way)
+    Long networkId;
+    String broadcastUri;
 
-    public VmWorkRemoveVmFromNetwork(long userId, long accountId, long vmId, String handlerName, Network network, URI broadcastUri) {
+    public VmWorkRemoveVmFromNetwork(long userId, long accountId, long vmId, String handlerName, long networkId, URI broadcastUri) {
         super(userId, accountId, vmId, handlerName);
 
-        this.network = network;
-        this.broadcastUri = broadcastUri;
+        this.networkId = networkId;
+        this.broadcastUri = broadcastUri != null ? broadcastUri.toString() : null;
     }
 
-    public VmWorkRemoveVmFromNetwork(VmWork vmWork, Network network, URI broadcastUri) {
+    public VmWorkRemoveVmFromNetwork(VmWork vmWork, long networkId, URI broadcastUri) {
         super(vmWork);
-        this.network = network;
-        this.broadcastUri = broadcastUri;
+        this.networkId = networkId;
+        this.broadcastUri = broadcastUri != null ? broadcastUri.toString() : null;
     }
 
-    public Network getNetwork() {
-        return network;
+    public Long getNetworkId() {
+        return networkId;
     }
 
     public URI getBroadcastUri() {
-        return broadcastUri;
+        return broadcastUri != null ? URI.create(broadcastUri) : null;
     }
 }