apache/cloudstack
1
0
Fork 0
mirror of https://github.com/apache/cloudstack.git synced 2025-10-26 08:42:29 +01:00
Code Issues Packages Projects Releases Wiki Activity

parallel nic adding (#5541)

Browse Source 
* trace nics additions

* work queue patch for network to add

* add secondary key to job

* logging improvements and naming of field(s)

* several naming corrections

* extra check if net already exists for vm

* placeholder job with secondary object

* constraint on entering the same job multiple times

* error handling/warning message

* review comments applied

Co-authored-by: Daan Hoogland <dahn@onecht.net>
Co-authored-by: Wei Zhou <wei.zhou@shapeblue.com>
This commit is contained in:
dahn 2021-10-04 17:21:21 +02:00 committed by GitHub
parent 93c0b602db
commit 3574d8d20b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 129 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java
View File

@ -42,6 +42,7 @@ import java.util.concurrent.TimeUnit;
import javax.inject.Inject;
import javax.naming.ConfigurationException;
import javax.persistence.EntityExistsException;
import org.apache.cloudstack.affinity.dao.AffinityGroupVMMapDao;
import org.apache.cloudstack.annotation.AnnotationService;
@ -3981,7 +3982,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
if (jobContext.isJobDispatchedBy(VmWorkConstants.VM_WORK_JOB_DISPATCHER)) {
// avoid re-entrance
VmWorkJobVO placeHolder = null;
placeHolder = createPlaceHolderWork(vm.getId());
placeHolder = createPlaceHolderWork(vm.getId(), network.getUuid());
try {
return orchestrateAddVmToNetwork(vm, network, requested);
} finally {
@ -4021,10 +4022,23 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
}
}
/**
* duplicated in {@see UserVmManagerImpl} for a {@see UserVmVO}
*/
private void checkIfNetworkExistsForVM(VirtualMachine virtualMachine, Network network) {
List<NicVO> allNics = _nicsDao.listByVmId(virtualMachine.getId());
for (NicVO nic : allNics) {
if (nic.getNetworkId() == network.getId()) {
throw new CloudRuntimeException("A NIC already exists for VM:" + virtualMachine.getInstanceName() + " in network: " + network.getUuid());
}
}
}
private NicProfile orchestrateAddVmToNetwork(final VirtualMachine vm, final Network network, final NicProfile requested) throws ConcurrentOperationException, ResourceUnavailableException,
InsufficientCapacityException {
final CallContext cctx = CallContext.current();
checkIfNetworkExistsForVM(vm, network);
s_logger.debug("Adding vm " + vm + " to network " + network + "; requested nic profile " + requested);
final VMInstanceVO vmVO = _vmDao.findById(vm.getId());
final ReservationContext context = new ReservationContextImpl(null, null, cctx.getCallingUser(), cctx.getCallingAccount());
@ -5375,7 +5389,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
Map<Volume, StoragePool> volumeStorageMap = dest.getStorageForDisks();
if (volumeStorageMap != null) {
for (Volume vol : volumeStorageMap.keySet()) {
checkConcurrentJobsPerDatastoreThreshhold(volumeStorageMap.get(vol));
checkConcurrentJobsPerDatastoreThreshold(volumeStorageMap.get(vol));
}
}
@ -5540,7 +5554,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
return new VmJobVirtualMachineOutcome(workJob, vm.getId());
}
private void checkConcurrentJobsPerDatastoreThreshhold(final StoragePool destPool) {
private void checkConcurrentJobsPerDatastoreThreshold(final StoragePool destPool) {
final Long threshold = VolumeApiService.ConcurrentMigrationsThresholdPerDatastore.value();
if (threshold != null && threshold > 0) {
long count = _jobMgr.countPendingJobs("\"storageid\":\"" + destPool.getUuid() + "\"", MigrateVMCmd.class.getName(), MigrateVolumeCmd.class.getName(), MigrateVolumeCmdByAdmin.class.getName());
@ -5561,7 +5575,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
Set<Long> uniquePoolIds = new HashSet<>(poolIds);
for (Long poolId : uniquePoolIds) {
StoragePoolVO pool = _storagePoolDao.findById(poolId);
checkConcurrentJobsPerDatastoreThreshhold(pool);
checkConcurrentJobsPerDatastoreThreshold(pool);
}
final VMInstanceVO vm = _vmDao.findByUuid(vmUuid);
@ -5608,14 +5622,34 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
final List<VmWorkJobVO> pendingWorkJobs = _workJobDao.listPendingWorkJobs(
VirtualMachine.Type.Instance, vm.getId(),
VmWorkAddVmToNetwork.class.getName());
VmWorkAddVmToNetwork.class.getName(), network.getUuid());
VmWorkJobVO workJob = null;
if (pendingWorkJobs != null && pendingWorkJobs.size() > 0) {
assert pendingWorkJobs.size() == 1;
if (pendingWorkJobs.size() > 1) {
s_logger.warn(String.format("The number of jobs to add network %s to vm %s are %d", network.getUuid(), vm.getInstanceName(), pendingWorkJobs.size()));
}
workJob = pendingWorkJobs.get(0);
} else {
if (s_logger.isTraceEnabled()) {
s_logger.trace(String.format("no jobs to add network %s for vm %s yet", network, vm));
}
workJob = createVmWorkJobToAddNetwork(vm, network, requested, context, user, account);
}
AsyncJobExecutionContext.getCurrentExecutionContext().joinJob(workJob.getId());
return new VmJobVirtualMachineOutcome(workJob, vm.getId());
}
private VmWorkJobVO createVmWorkJobToAddNetwork(
VirtualMachine vm,
Network network,
NicProfile requested,
CallContext context,
User user,
Account account) {
VmWorkJobVO workJob;
workJob = new VmWorkJobVO(context.getContextId());
workJob.setDispatcher(VmWorkConstants.VM_WORK_JOB_DISPATCHER);
@ -5626,17 +5660,23 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
workJob.setVmType(VirtualMachine.Type.Instance);
workJob.setVmInstanceId(vm.getId());
workJob.setRelated(AsyncJobExecutionContext.getOriginJobId());
workJob.setSecondaryObjectIdentifier(network.getUuid());
// save work context info (there are some duplications)
final VmWorkAddVmToNetwork workInfo = new VmWorkAddVmToNetwork(user.getId(), account.getId(), vm.getId(),
VirtualMachineManagerImpl.VM_WORK_JOB_HANDLER, network.getId(), requested);
workJob.setCmdInfo(VmWorkSerializer.serialize(workInfo));
try {
_jobMgr.submitAsyncJob(workJob, VmWorkConstants.VM_WORK_QUEUE, vm.getId());
} catch (CloudRuntimeException e) {
if (e.getCause() instanceof EntityExistsException) {
String msg = String.format("A job to add a nic for network %s to vm %s already exists", network.getUuid(), vm.getUuid());
s_logger.warn(msg, e);
}
AsyncJobExecutionContext.getCurrentExecutionContext().joinJob(workJob.getId());
return new VmJobVirtualMachineOutcome(workJob, vm.getId());
throw e;
}
return workJob;
}
public Outcome<VirtualMachine> removeNicFromVmThroughJobQueue(
@ -5945,6 +5985,10 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
}
private VmWorkJobVO createPlaceHolderWork(final long instanceId) {
return createPlaceHolderWork(instanceId, null);
}
private VmWorkJobVO createPlaceHolderWork(final long instanceId, String secondaryObjectIdentifier) {
final VmWorkJobVO workJob = new VmWorkJobVO("");
workJob.setDispatcher(VmWorkConstants.VM_WORK_JOB_PLACEHOLDER);
@ -5956,6 +6000,9 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
workJob.setStep(VmWorkJobVO.Step.Starting);
workJob.setVmType(VirtualMachine.Type.Instance);
workJob.setVmInstanceId(instanceId);
if(StringUtils.isNotBlank(secondaryObjectIdentifier)) {
workJob.setSecondaryObjectIdentifier(secondaryObjectIdentifier);
}
workJob.setInitMsid(ManagementServerNode.getManagementServerId());
_workJobDao.persist(workJob);

3
engine/schema/src/main/resources/META-INF/db/schema-41520to41600.sql
View File

@ -791,3 +791,6 @@ ALTER TABLE cloud.user_vm_details MODIFY value varchar(5120) NOT NULL;
ALTER TABLE cloud_usage.usage_network DROP PRIMARY KEY, ADD PRIMARY KEY (`account_id`,`zone_id`,`host_id`,`network_id`,`event_time_millis`);
ALTER TABLE `cloud`.`user_statistics` DROP INDEX `account_id`, ADD UNIQUE KEY `account_id` (`account_id`,`data_center_id`,`public_ip_address`,`device_id`,`device_type`, `network_id`);
ALTER TABLE `cloud_usage`.`user_statistics` DROP INDEX `account_id`, ADD UNIQUE KEY `account_id` (`account_id`,`data_center_id`,`public_ip_address`,`device_id`,`device_type`, `network_id`);
ALTER TABLE `cloud`.`vm_work_job` ADD COLUMN `secondary_object` char(100) COMMENT 'any additional item that must be checked during queueing' AFTER `vm_instance_id`;
ALTER TABLE cloud.vm_work_job ADD CONSTRAINT vm_work_job_step_and_objects UNIQUE KEY (step,vm_instance_id,secondary_object);

2
framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/dao/VmWorkJobDao.java
View File

@ -32,6 +32,8 @@ public interface VmWorkJobDao extends GenericDao<VmWorkJobVO, Long> {
List<VmWorkJobVO> listPendingWorkJobs(VirtualMachine.Type type, long instanceId, String jobCmd);
List<VmWorkJobVO> listPendingWorkJobs(VirtualMachine.Type type, long instanceId, String jobCmd, String secondaryObjectIdentifier);
void updateStep(long workJobId, Step step);
void expungeCompletedWorkJobs(Date cutDate);

15
framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/dao/VmWorkJobDaoImpl.java
View File

@ -67,6 +67,7 @@ public class VmWorkJobDaoImpl extends GenericDaoBase<VmWorkJobVO, Long> implemen
PendingWorkJobByCommandSearch.and("jobStatus", PendingWorkJobByCommandSearch.entity().getStatus(), Op.EQ);
PendingWorkJobByCommandSearch.and("vmType", PendingWorkJobByCommandSearch.entity().getVmType(), Op.EQ);
PendingWorkJobByCommandSearch.and("vmInstanceId", PendingWorkJobByCommandSearch.entity().getVmInstanceId(), Op.EQ);
PendingWorkJobByCommandSearch.and("secondaryObjectIdentifier", PendingWorkJobByCommandSearch.entity().getSecondaryObjectIdentifier(), Op.EQ);
PendingWorkJobByCommandSearch.and("step", PendingWorkJobByCommandSearch.entity().getStep(), Op.NEQ);
PendingWorkJobByCommandSearch.and("cmd", PendingWorkJobByCommandSearch.entity().getCmd(), Op.EQ);
PendingWorkJobByCommandSearch.done();
@ -119,6 +120,20 @@ public class VmWorkJobDaoImpl extends GenericDaoBase<VmWorkJobVO, Long> implemen
return this.listBy(sc, filter);
}
@Override
public List<VmWorkJobVO> listPendingWorkJobs(VirtualMachine.Type type, long instanceId, String jobCmd, String secondaryObjectIdentifier) {
SearchCriteria<VmWorkJobVO> sc = PendingWorkJobByCommandSearch.create();
sc.setParameters("jobStatus", JobInfo.Status.IN_PROGRESS);
sc.setParameters("vmType", type);
sc.setParameters("vmInstanceId", instanceId);
sc.setParameters("secondaryObjectIdentifier", secondaryObjectIdentifier);
sc.setParameters("cmd", jobCmd);
Filter filter = new Filter(VmWorkJobVO.class, "created", true, null, null);
return this.listBy(sc, filter);
}
@Override
public void updateStep(long workJobId, Step step) {
VmWorkJobVO jobVo = findById(workJobId);

2
framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobVO.java
View File

@ -384,7 +384,7 @@ public class AsyncJobVO implements AsyncJob, JobInfo {
@Override
public String toString() {
StringBuffer sb = new StringBuffer();
sb.append("AsyncJobVO {id:").append(getId());
sb.append("AsyncJobVO : {id:").append(getId());
sb.append(", userId: ").append(getUserId());
sb.append(", accountId: ").append(getAccountId());
sb.append(", instanceType: ").append(getInstanceType());

24
framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/VmWorkJobVO.java
View File

@ -58,6 +58,9 @@ public class VmWorkJobVO extends AsyncJobVO {
@Column(name = "vm_instance_id")
long vmInstanceId;
@Column(name = "secondary_object")
String secondaryObjectIdentifier;
protected VmWorkJobVO() {
}
@ -89,4 +92,25 @@ public class VmWorkJobVO extends AsyncJobVO {
public void setVmInstanceId(long vmInstanceId) {
this.vmInstanceId = vmInstanceId;
}
public String getSecondaryObjectIdentifier() {
return secondaryObjectIdentifier;
}
public void setSecondaryObjectIdentifier(String secondaryObjectIdentifier) {
this.secondaryObjectIdentifier = secondaryObjectIdentifier;
}
@Override
public String toString() {
StringBuffer sb = new StringBuffer();
sb.append("VmWorkJobVO : {").
append(", step: ").append(getStep()).
append(", vmType: ").append(getVmType()).
append(", vmInstanceId: ").append(getVmInstanceId()).
append(", secondaryObjectIdentifier: ").append(getSecondaryObjectIdentifier()).
append(super.toString()).
append("}");
return sb.toString();
}
}

21
server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
View File

@ -1384,12 +1384,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
Account vmOwner = _accountMgr.getAccount(vmInstance.getAccountId());
_networkModel.checkNetworkPermissions(vmOwner, network);
List<NicVO> allNics = _nicDao.listByVmId(vmInstance.getId());
for (NicVO nic : allNics) {
if (nic.getNetworkId() == network.getId()) {
throw new CloudRuntimeException("A NIC already exists for VM:" + vmInstance.getInstanceName() + " in network: " + network.getUuid());
}
}
checkIfNetExistsForVM(vmInstance, network);
macAddress = validateOrReplaceMacAddress(macAddress, network.getId());
@ -1456,10 +1451,22 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
}
}
CallContext.current().putContextParameter(Nic.class, guestNic.getUuid());
s_logger.debug("Successful addition of " + network + " from " + vmInstance);
s_logger.debug(String.format("Successful addition of %s from %s through %s", network, vmInstance, guestNic));
return _vmDao.findById(vmInstance.getId());
}
/**
* duplicated in {@see VirtualMachineManagerImpl} for a {@see VMInstanceVO}
*/
private void checkIfNetExistsForVM(VirtualMachine virtualMachine, Network network) {
List<NicVO> allNics = _nicDao.listByVmId(virtualMachine.getId());
for (NicVO nic : allNics) {
if (nic.getNetworkId() == network.getId()) {
throw new CloudRuntimeException("A NIC already exists for VM:" + virtualMachine.getInstanceName() + " in network: " + network.getUuid());
}
}
}
/**
* If the given MAC address is invalid it replaces the given MAC with the next available MAC address
*/