...

1. Background

VM life cycle in CloudStack is current represented through a number of lifecycle VM states, following is a complete list of these states

 Starting, Running, Stopping, Stopped, Destroyed, Migrating, Expunging, Error, Unknown

...

Upon hypervisor host-connect event, hypervisor resource-agent will first report all VMs on the host to management server, it triggers a "full-sync" process with management server to build an initial sync start point, the host won't be considered as in UP state until this "full-sync" process is completed. After host is connected, hypervisor host resource-agent will periodically perform "delta-sync" with CloudStack management server.

 "Full-sync" and "delta-sync" are currently forming the foundation of VMSync process in CloudStack. Although it works nicely most of time for VMs that are solely operated by CloudStack, as soon as the introduction of external VM managers are involved, for example Citrix XenCenter, VMware vCenter, the state sync scenarios can become hard to handle when out-of-band changes posted from external managers, following use cases sometimes can cause problematic issues during normal operations of CloudStack.

...

This improvement effort is to address these issues, it will help CloudStack to better interage with third-party virtualization managers like VMware vCenter to perform HA, DRS, FT etc better and reliably through CloudStack.

...

2. Design

 2.1High1 High-level principals

At very high-level, we try to attack the problem in following areas

...

With VM power state, hypervisor resource-agent no longer needs to know anything about a transition job status that is specific to CloudStack, all it needs to care is how to carry on a hypervisor-specific action or report VM power state periodically, there is no need to setup a sync start point, therefore we can eliminate "full-sync" process at all.

Following is a code snaplet that shows the old way of how a hypervisor resource-agent needs to do a sync report

...

2) Serialize VM operations

Currently, handling of state transition handling always happens at in-place context, for example, when management server receives hypervisor VM state report, the handling of the report is processed within the context, even if there may be another thread that is handling user request on the same VM. Although we try to coordinate by checking the state of the VM, by simplify failing it with concurrent-access exception.  

About in-place handling style, following code snaplet shows an example.

     
    protected Command compareState(long hostId, VMInstanceVO vm, final AgentVmInfo info, final boolean fullSync, boolean trackExternalChange) {
        State agentState = info.state;
        final State serverState = vm.getState();
        final String serverName = vm.getInstanceName();

        Command command = null;
        s_logger.debug("VM " + serverName + ": cs state = " + serverState + " and realState = " + agentState);
        if (s_logger.isDebugEnabled()) {
            s_logger.debug("VM " + serverName + ": cs state = " + serverState + " and realState = " + agentState);
        }

        if (agentState == State.Error) {
            agentState = State.Stopped;

            short alertType = AlertManager.ALERT_TYPE_USERVM;
            if (VirtualMachine.Type.DomainRouter.equals(vm.getType())) {
                alertType = AlertManager.ALERT_TYPE_DOMAIN_ROUTER;
            } else if (VirtualMachine.Type.ConsoleProxy.equals(vm.getType())) {
                alertType = AlertManager.ALERT_TYPE_CONSOLE_PROXY;
            } else if (VirtualMachine.Type.SecondaryStorageVm.equals(vm.getType())) {
                alertType = AlertManager.ALERT_TYPE_SSVM;
            }

            HostPodVO podVO = _podDao.findById(vm.getPodIdToDeployIn());
            DataCenterVO dcVO = _dcDao.findById(vm.getDataCenterId());
            HostVO hostVO = _hostDao.findById(vm.getHostId());

            String hostDesc = "name: " + hostVO.getName() + " (id:" + hostVO.getId() + "), availability zone: " + dcVO.getName() + ", pod: " + podVO.getName();
            _alertMgr.sendAlert(alertType, vm.getDataCenterId(), vm.getPodIdToDeployIn(), "VM (name: " + vm.getInstanceName() + ", id: " + vm.getId() + ") stopped on host " + hostDesc
                    + " due to storage failure", "Virtual Machine " + vm.getInstanceName() + " (id: " + vm.getId() + ") running on host [" + vm.getHostId() + "] stopped due to storage failure.");
        }

        if (trackExternalChange) {
            if (serverState == State.Starting) {
                if (vm.getHostId() != null && vm.getHostId() != hostId) {
                    s_logger.info("CloudStack is starting VM on host " + vm.getHostId() + ", but status report comes from a different host " + hostId + ", skip status sync for vm: "
                            + vm.getInstanceName());
                    return null;
                }
            }
            if (vm.getHostId() == null || hostId != vm.getHostId()) {
                try {
                    ItWorkVO workItem = _workDao.findByOutstandingWork(vm.getId(), State.Migrating);
                    if(workItem == null){
                        stateTransitTo(vm, VirtualMachine.Event.AgentReportMigrated, hostId);
                    }
                } catch (NoTransitionException e) {
                }
            }
        }

        // during VM migration time, don't sync state will agent status update
        if (serverState == State.Migrating) {
            s_logger.debug("Skipping vm in migrating state: " + vm);
            return null;
        }

        if (trackExternalChange) {
            if (serverState == State.Starting) {
                if (vm.getHostId() != null && vm.getHostId() != hostId) {
                    s_logger.info("CloudStack is starting VM on host " + vm.getHostId() + ", but status report comes from a different host " + hostId + ", skip status sync for vm: "
                            + vm.getInstanceName());
                    return null;
                }
            }

            if (serverState == State.Running) {
                try {
                    //
                    // we had a bug that sometimes VM may be at Running State
                    // but host_id is null, we will cover it here.
                    // means that when CloudStack DB lost of host information,
                    // we will heal it with the info reported from host
                    //
                    if (vm.getHostId() == null || hostId != vm.getHostId()) {
                        if (s_logger.isDebugEnabled()) {
                            s_logger.debug("detected host change when VM " + vm + " is at running state, VM could be live-migrated externally from host " + vm.getHostId() + " to host " + hostId);
                        }

                        stateTransitTo(vm, VirtualMachine.Event.AgentReportMigrated, hostId);
                    }
                } catch (NoTransitionException e) {
                    s_logger.warn(e.getMessage());
                }
            }
        }

        if (agentState == serverState) {
            if (s_logger.isDebugEnabled()) {
                s_logger.debug("Both states are " + agentState + " for " + vm);
            }
            assert (agentState == State.Stopped || agentState == State.Running) : "If the states we send up is changed, this must be changed.";
            if (agentState == State.Running) {
                try {
                    stateTransitTo(vm, VirtualMachine.Event.AgentReportRunning, hostId);
                } catch (NoTransitionException e) {
                    s_logger.warn(e.getMessage());
                }
                // FIXME: What if someone comes in and sets it to stopping? Then
                // what?
                return null;
            }

            s_logger.debug("State matches but the agent said stopped so let's send a cleanup command anyways.");
            return cleanup(vm);
        }

        if (agentState == State.Shutdowned) {
            if (serverState == State.Running || serverState == State.Starting || serverState == State.Stopping) {
                try {
                    advanceStop(vm, true, _accountMgr.getSystemUser(), _accountMgr.getSystemAccount());
                } catch (AgentUnavailableException e) {
                    assert (false) : "How do we hit this with forced on?";
                    return null;
                } catch (OperationTimedoutException e) {
                    assert (false) : "How do we hit this with forced on?";
                    return null;
                } catch (ConcurrentOperationException e) {
                    assert (false) : "How do we hit this with forced on?";
                    return null;
                }
            } else {
                s_logger.debug("Sending cleanup to a shutdowned vm: " + vm.getInstanceName());
                command = cleanup(vm);
            }
        } else if (agentState == State.Stopped) {
            // This state means the VM on the agent was detected previously
            // and now is gone. This is slightly different than if the VM
            // was never completed but we still send down a Stop Command
            // to ensure there's cleanup.
            if (serverState == State.Running) {
                // Our records showed that it should be running so let's restart
                // it.
                _haMgr.scheduleRestart(vm, false);
            } else if (serverState == State.Stopping) {
                _haMgr.scheduleStop(vm, hostId, WorkType.ForceStop);
                s_logger.debug("Scheduling a check stop for VM in stopping mode: " + vm);
            } else if (serverState == State.Starting) {
                s_logger.debug("Ignoring VM in starting mode: " + vm.getInstanceName());
                _haMgr.scheduleRestart(vm, false);
            }
            command = cleanup(vm);
        } else if (agentState == State.Running) {
            if (serverState == State.Starting) {
                if (fullSync) {
                    try {
                        ensureVmRunningContext(hostId, vm, Event.AgentReportRunning);
                    } catch (OperationTimedoutException e) {
                        s_logger.error("Exception during update for running vm: " + vm, e);
                        return null;
                    } catch (ResourceUnavailableException e) {
                        s_logger.error("Exception during update for running vm: " + vm, e);
                        return null;
                    }catch (InsufficientAddressCapacityException e) {
                        s_logger.error("Exception during update for running vm: " + vm, e);
                        return null;
                    }catch (NoTransitionException e) {
                        s_logger.warn(e.getMessage());
                    }
                }
            } else if (serverState == State.Stopping) {
                s_logger.debug("Scheduling a stop command for " + vm);
                _haMgr.scheduleStop(vm, hostId, WorkType.Stop);
            } else {
                s_logger.debug("server VM state " + serverState + " does not meet expectation of a running VM report from agent");

                // just be careful not to stop VM for things we don't handle
                // command = cleanup(vm);
            }
        }
        return command;
    }

...

Follow code snaplet gives a synchronized synchronous handling logic example that is supported in the new model.

    @Override
    public <T extends VMInstanceVO> boolean advanceStop(final T vm, boolean forced, User user, Account account) throws AgentUnavailableException, OperationTimedoutException, ConcurrentOperationException {
    	VmWorkJobVO workJob = null;
    	Transaction txn = Transaction.currentTxn();
    	try {
        	txn.start();

        	_vmDao.lockRow(vm.getId(), true);

        	List<VmWorkJobVO> pendingWorkJobs = _workJobDao.listPendingWorkJobs(
        		VirtualMachine.Type.Instance, vm.getId(), VmWorkConstants.VM_WORK_STOP);

        	if(pendingWorkJobs != null && pendingWorkJobs.size() > 0) {
        		assert(pendingWorkJobs.size() == 1);
        		workJob = pendingWorkJobs.get(0);
        	} else {
        		workJob = new VmWorkJobVO();

        		workJob.setDispatcher(VmWorkConstants.VM_WORK_JOB_DISPATCHER);
        		workJob.setCmd(VmWorkConstants.VM_WORK_STOP);

        		workJob.setAccountId(account.getId());
        		workJob.setUserId(user.getId());
        		workJob.setStep(VmWorkJobVO.Step.Prepare);
        		workJob.setVmType(vm.getType());
        		workJob.setVmInstanceId(vm.getId());

        		// save work context info (there are some duplications)
        		VmWorkStop workInfo = new VmWorkStop();
        		workInfo.setAccountId(account.getId());
        		workInfo.setUserId(user.getId());
        		workInfo.setVmId(vm.getId());
        		workInfo.setForceStop(forced);
        		workJob.setCmdInfo(ApiSerializerHelper.toSerializedString(workInfo));

        		_jobMgr.submitAsyncJob(workJob, VmWorkConstants.VM_WORK_QUEUE, vm.getId());
        	}

        	txn.commit();
    	} catch(Throwable e) {
    		s_logger.error("Unexpected exception", e);
    		txn.rollback();
    		throw new ConcurrentOperationException("Unhandled exception, converted to ConcurrentOperationException");
    	}

    	final long jobId = workJob.getId();
    	AsyncJobExecutionContext.getCurrentExecutionContext().joinJob(jobId);

    	//
    	// TODO : this will be replaced with fully-asynchronizedasynchronous way later so that we don't need
    	// to wait here. The reason we do it synchronizedsynchronous here is that callers of advanceStart is expecting
    	// synchronizedsynchronous semantics
    	//
    	//
    	_jobMgr.waitAndCheck(
    		new String[] { TopicConstants.VM_POWER_STATE, TopicConstants.JOB_STATE },
    		3000L, 600000L, new Predicate() {

				@Override
				public boolean checkCondition() {
					VMInstanceVO instance = _vmDao.findById(vm.getId());
					if(instance.getPowerState() == VirtualMachine.PowerState.PowerOff)
						return true;

					VmWorkJobVO workJob = _workJobDao.findById(jobId);
					if(workJob.getStatus() != AsyncJobConstants.STATUS_IN_PROGRESS)
						return true;

					return false;
				}
    		});

    	try {
    		AsyncJobExecutionContext.getCurrentExecutionContext().disjoinJob(jobId);
    	} catch(Exception e) {
    		s_logger.error("Unexpected exception", e);
    		return false;
    	}
    	return true;
    }

...

All subscribers on agent related topic (i.e., VirtualMachineManagerImpl, ConsoleProxyManagerImpl etc) have to explicitly wire to the publisher (AgentManagerImpl), as the needs of different topic grows, these explicit wirings make the whole system tightly coupled between components.  

2.2Code 2 Code changes

2.2.1 Message Bus facility

...

 
MessageBus defines the interface of the message bus facility, it implements a simple publish/subscribe pattern, publishers and subscribers can be linked by sharing a common topic,  topic can be in hierarchy mode, a subscriber at higher hierarchy mode can receive messages from all topics that are below. 

...

_jobMgr.waitAndCheck relies on the service provided by MessageDetector. To fit into synchronized programming model, MessageDetector will block the execution until an interested event appears on the message bus.

2.2.2 Job facility

Job types

TopicConstants

public interface TopicConstants {
	// VM power state messages on message bus
	public static final String VM_POWER_STATE = "vm.powerstate";

	// job messages on message bus
	public static final String JOB_HEARTBEAT = "job.heartbeat";
	public static final String JOB_STATE = "job.state";


	...
}

In loosely-coupled model, topics on the message bus become the glue point between components, to avoid the coupling become too loose, all topics on the message bus are put into central management. Topics and related parameters with the topic needs to be defined in TopicConstants. 

2.2.2 Job facility

Job types

A job in CloudStack actually represents an orchestration work flow. Due to historic reason, CloudStack has taken considerable efforts trying to make the concept of job implicit to programmers, this is done through the API Command pattern, for API command that is executed asynchronously, the request will first be posted to an internal job facility A job in CloudStack actually represents an orchestration work flow. Due to historic reason, CloudStack has taken considerable efforts trying to make the concept of job implicit to programmers, this is done through the API Command pattern, for API command that is executed asynchronizedly, the request will first be posted to an internal job facility but real execution/processing will be called back into the command object from within the job thread context. The whole job facility has been made implicit intentionally, in most of cases, job facility is used as a context switcher to just provide the execution thread context. This implicit use of job facility actually treats job as secondary class, since explicit job control is discouraged, it leads to the programming model to handle things in-place within the calling context, synchronization is then usually done through locking. 

In this refactoring proposal, we will promote jobs into first-class objects, jobs are encouraged to be used in a more explicit way. We will use ordered orderly execution to help reduce the use of locking across the code base and manage orchestration processes explicitly. This will give us better control on managing system load, knows when we need to scale management server cluster and no longer in mercy of java thread pools.

However, due to the legacy bagage we have, moving to this direction will be a long journey, in order to make existing model work without too much change, instead of having one ideal abstract job facility that does general job scheduling, execution and ordering, we will have 3 major job types currently.

1) API job

2) Work job

3) Pseduo Job

AsyncJobManagerImpl

Refactor it to decouple the tight link with API jobs, make it generic not only executing async API request jobs but also executing internal VM operating jobs

ApiAsyncJobDispatcher

Dispatch async API request jobs

VmWorkJobDispatcher

dispatch internal async VM operation jobs

VmWorkJobVO

VmWorkJobDao

VmWorkJobDaoImpl

Persist classes for internal VM operation jobs

AsyncJobJournalVO

AsyncJobJournalDao

AsyncJobJournalDaoImpl 

Implements job journal facility, all jobs can now have a persist job journal facility

2.2.3 VM Power state sync

VirtualMachinePowerStateSync

VirtualMachinePowerStateSyncImpl

2.2.4 Other refactored classes

VirtualMachineManagerImpl

HighAvailabilityManagerImpl

VirtualMachineGuru

ReservationContext

Hypervisor resource classes

etc.

...

API job gives a running context for an asynchronous API request, it usually starts an orchestration process. 

2) Work job

Work job in the new model carries the real orchestrator process, its run will be serialized on the target VM object.

3) Pseudo Job

Inside CloudStack, there are a few manager components that use their own threads to manage service activities, when it comes to use the newly introduced work jobs for orchestration, we sometimes need a pseudo job context,  pseudo job provides just that context. The difference between Pseudo job and an high level API job is that pseudo job runs in its own thread context, while API job runs in the thread from job thread pool.

Job joining

Like a process in operating system, a job can have multiple execution states, it could be put in blocking, or be in currently running state, etc. Joining another job means to wait for completion of the subject job, be either a successful completion or a failure completion. A blocked job may be rescheduled to run based on triggering of events. Currently, a job that is joining to another job can only be scheduled to run upon wakeup events.

Job wakeup

When a job joins to another job, to wait for the completion status of joined job, there are two ways to achieve that. We've shown it for the first way in _jobMgr.waitAndCheck - blocking the executing thread until the condition is satisfied. The problem of this approach is that it holds a real executing thread. If a caller already has a persistent thread, it is not a problem, however, for most of API initiated orchestration jobs, they all share a global job thread pool, blocking executing thread is not the most efficient way for system scalability. The new job facility provides a support for a second approach, this approach will put the job into blocking state, release the executing thread, and then reschedule job execution based on wakeup calls(event triggered).

Message bus provides delivery-at-best service to CloudStack components, for locally broadcast messages(within one management server), it is reliable when management server is running, however, for messages that are across-ing management server boundaries, it is not a 100 percent reliable service for building a reliable orchestration process. When a job is joining and waiting for another job to complete, in order for job check-up process keep on going, the job will be periodically waken up on a specified interval. Therefore, message bus service can be used for efficient event notification and in case that message bus service fails, this wakeup service will help ensure the reliability of the whole orchestration process.

Job heartbeat and Job monitoring

When a job is in executing, to help better monitor system health and load, we've added a Job monitoring facility, to track all active job threads and job heartbeats. All cooperative jobs are required to periodically report job heartbeats on the message bus with topic of TopicConstants.JOB_HEARTBEAT. Stalling jobs can be identified through this monitoring service and in the future we will add graceful job cancellation support if the job's active activity can be interrupted and cancelled.

AsyncJobManagerImpl

Refactor it to decouple the tight linkage with API jobs before, make it generic not only executing async API request jobs but also executing internal VM operating jobs

ApiAsyncJobDispatcher

Dispatch async API request jobs

VmWorkJobDispatcher

dispatch internal async VM operation jobs

VmWorkWakeupJobDispatcher

Dispatch wakeup event to blocking jobs

VmWorkJobVO/VmWorkJobDao/VmWorkJobDaoImpl

Persist classes for internal VM operation jobs

AsyncJobJournalVO/AsyncJobJournalDao/AsyncJobJournalDaoImpl 

Implements job journal facility, all jobs can now have a persist job journal facility

2.2.3 VM Power state sync

VirtualMachinePowerStateSync/VirtualMachinePowerStateSyncImpl

Strictly speaking, there will be no "Sync" fix in the new model at all. Since the whole previous "sync" process has become unnecessary in the new model. When a VM power state report is received by management server, all it does can be abstracted in following work flow

if(  there is a pending job on the subject VM ) {

	Raise wakeup call to the job

} else {

	Save the report info

        trigger and schedule responding job if neccessary

}

When there is no pending job working on the VM, VM should always stay at stationary states (i.e., PoweredOn or PoweredOff), out of sync situation between what CloudStack DB has record and what a host has reported will be resolved with a new job flow, depends on HA configuration, we can either try to eventually bring VM state to be in sync with CloudStack DB or let CloudStack honor what it is reported.

Every flow job starts from one stationary VM state and ends to another/same stationary state, the transition process will be solely conducted and monitored by the job itself, there is no need for a traditional "Sync" process to get involved. 

2.2.4 Other refactored classes

VirtualMachineManagerImpl

HighAvailabilityManagerImpl

VirtualMachineGuru

ReservationContext

Hypervisor resource classes

etc.

3. Use cases

To limit the scope of required refactoring work, we still keep legacy VM states as we have before, the difference is that for operations that could cause state transitions, it will always be managed solely from a job and as long as the management server is running, any transition process should either succeed or fail within a certain limit of time period. Following is the guideline to determine any expected system behave.

1) As long as management server is running, no VM should be stuck at a certain transitional state for too long, these transitional states include Starting, Stoppping, Migrating and Expunging.

2) For any transitional process, i.e., transition from Stopped -> Running through Starting state, if during the time there is external changes on the VM (i.e., stop a just started VM from vCenter), it can only affect the result of job, if the transition direction from external side is the same as what the job is expecting, it has no effect to the final result, otherwise, it can eventually cause the job to fail, in any way, upon the completion of the job, VM will be returned to a stationary state.

3) Upon management server restart, any pending un-completed jobs will be failed, related resources in CloudStack should be released and related VMs should be put into the last known stationary state.

4) If host reports a out-of-sync power state while there is no pending job that is acting on the VM, for non-HA case, CloudStack will always update VM to be in sync with what it is reported from hypervisor, however, since this state transition is not gone through a normal transitional job, and if VM is put into Running state, VM's network environment will most likely be not ready, we will report an system alert about this VM for manual attention from administrator or end-user. However, since the VM has been put into a stationary state that is in-sync with what hypervisor has reported, CloudStack or end-user can always issue an correction command to either stop or start the VM through normal process.

5) To support external VM live migration, a host change in host report will always be honored by CloudStack, regardless whether or not there is a pending job that is acting on the VM. To honor host change of VM, CloudStack updates the VM/host relationship accordingly. Considering a complex migration case as below,

A VM-migrate transition job is trying to lively migrate VM from host 1 to host 2, however, during the time, if external manager like vCenter interrupts it and has migrated it from host 1 to host 3 successfully, the VM-migrating job in CloudStack should fail as the end result of the VM state ends up at an unexpected host, nevertheless, upon completion of this failed job, VM's host change will still be honored so that user can re-perform the migration again, but this time, command will be issued from host 3 to host 2. 

4. Schema changes

ALTER TABLE `cloud`.`async_job` DROP COLUMN `session_key`;

ALTER TABLE `cloud`.`async_job` DROP COLUMN `job_cmd_originator`;

ALTER TABLE `cloud`.`async_job` DROP COLUMN `callback_type`;

ALTER TABLE `cloud`.`async_job` DROP COLUMN `callback_address`;




ALTER TABLE `cloud`.`async_job` ADD COLUMN `job_type` VARCHAR(32);

ALTER TABLE `cloud`.`async_job` ADD COLUMN `job_dispatcher` VARCHAR(64);

ALTER TABLE `cloud`.`async_job` ADD COLUMN `job_executing_msid` bigint;

ALTER TABLE `cloud`.`async_job` ADD COLUMN `job_pending_signals` int(10) NOT NULL DEFAULT 0;




ALTER TABLE `cloud`.`vm_instance` ADD COLUMN `power_state` VARCHAR(64) DEFAULT 'PowerUnknown';

ALTER TABLE `cloud`.`vm_instance` ADD COLUMN `power_state_update_time` DATETIME;

ALTER TABLE `cloud`.`vm_instance` ADD COLUMN `power_state_update_count` INT DEFAULT 0;

ALTER TABLE `cloud`.`vm_instance` ADD COLUMN `power_host` bigint unsigned;

ALTER TABLE `cloud`.`vm_instance` ADD CONSTRAINT `fk_vm_instance__power_host` FOREIGN KEY (`power_host`) REFERENCES `cloud`.`host`(`id`);




CREATE TABLE `cloud`.`vm_work_job` (

   `id` bigint unsigned UNIQUE NOT NULL,

   `step` char(32) NOT NULL COMMENT 'state',

   `vm_type` char(32) NOT NULL COMMENT 'type of vm',

   `vm_instance_id` bigint unsigned NOT NULL COMMENT 'vm instance',

   PRIMARY KEY (`id`),

   CONSTRAINT `fk_vm_work_job__instance_id` FOREIGN KEY (`vm_instance_id`) REFERENCES `vm_instance`(`id`) ON DELETE CASCADE,

   INDEX `i_vm_work_job__vm`(`vm_type`, `vm_instance_id`),

   INDEX `i_vm_work_job__step`(`step`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;




CREATE TABLE `cloud`.`async_job_journal` (

   `id` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT 'id',

   `job_id` bigint unsigned NOT NULL,

   `journal_type` varchar(32),

   `journal_text` varchar(1024) COMMENT 'journal descriptive informaton',

   `journal_obj` varchar(1024) COMMENT 'journal strutural information, JSON encoded object',

   `created` datetime NOT NULL COMMENT 'date created',

   PRIMARY KEY (`id`),

   CONSTRAINT `fk_async_job_journal__job_id` FOREIGN KEY (`job_id`) REFERENCES `async_job`(`id`) ON DELETE CASCADE

) ENGINE=InnoDB DEFAULT CHARSET=utf8;




CREATE TABLE `cloud`.`async_job_join_map` (

   `id` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT 'id',

   `job_id` bigint unsigned NOT NULL,

   `join_job_id` bigint unsigned NOT NULL,

   `join_status` int NOT NULL,

   `join_result` varchar(1024),

   `join_msid` bigint,

   `complete_msid` bigint,

   `sync_source_id` bigint COMMENT 'upper-level job sync source info before join',

   `wakeup_handler` varchar(64),

   `wakeup_dispatcher` varchar(64),

   `wakeup_interval` bigint NOT NULL DEFAULT 3000 COMMENT 'wakeup interval in seconds',

   `created` datetime NOT NULL,

   `last_updated` datetime,

   `next_wakeup` datetime,

   `expiration` datetime,

   PRIMARY KEY (`id`),

   CONSTRAINT `fk_async_job_join_map__job_id` FOREIGN KEY (`job_id`) REFERENCES `async_job`(`id`) ON DELETE CASCADE,

   CONSTRAINT `fk_async_job_join_map__join_job_id` FOREIGN KEY (`join_job_id`) REFERENCES `async_job`(`id`),

   CONSTRAINT `fk_async_job_join_map__join` UNIQUE (`job_id`, `join_job_id`),

   INDEX `i_async_job_join_map__join_job_id`(`join_job_id`),

   INDEX `i_async_job_join_map__created`(`created`),

   INDEX `i_async_job_join_map__last_updated`(`last_updated`),

   INDEX `i_async_job_join_map__next_wakeup`(`next_wakeup`),

   INDEX `i_async_job_join_map__expiration`(`expiration`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

4. API/UI

This is low-level change that should keep API compatible, UI change is also not mandatory, we can have UI change to take advantage of better job management in the future(i.e. job journal for more descriptive error messages)