We will try to use a message-bus to co-ordinate different activities within the management server. This facility is different with the existing feature of "Event Bus", the later one is mainly to integrate external systems through persist-able message-queue servers.

Code changes

...

Message bus is a loosely-coupled way for publish/subscribe pattern. We currently use publish/subscribe pattern a lot, but in a strong-type way, for example, Listener interface

Code Block

public interface MessageBusListener {

void setMessageSerializer(MessageSerializer messageSerializer);

MessageSerializer getMessageSerializer();


void subscribe(String topic, MessageSubscriber subscriber);

void unsubscribe(String topic, MessageSubscriber subscriber);

void clearAll();

void prune();


void publish(String senderAddress, String topic, PublishScope scope, Object args);

}

MessageBus defines the interface of the message bus facility, it implements a simple publish/subscribe pattern, publishers and subscribers can 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.

MessageBusBase

A simple message bus implementation

MessageHandler

Java annotation for subscriber to specify a message handler

MessageDispatcher

For message subscriber to use to dispatch received messages to annotated message handlers

MessageDetector

To detect interested messages on message bus

2) Job facility

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

3) VM Power state sync

VirtualMachinePowerStateSync

VirtualMachinePowerStateSyncImpl

4) Other refactored classes

VirtualMachineManagerImpl

HighAvailabilityManagerImpl

VirtualMachineGuru

ReservationContext

Hypervisor resource classes

etc.

Schema changes

    boolean processAnswers(long agentId, long seq, Answer[] answers);
    boolean processCommands(long agentId, long seq, Command[] commands);
    AgentControlAnswer processControlCommand(long agentId, AgentControlCommand cmd);
    void processConnect(HostVO host, StartupCommand cmd, boolean forRebalance) throws ConnectionException;
    boolean processDisconnect(long agentId, Status state);
    boolean isRecurring();
    int getTimeout();
    boolean processTimeout(long agentId, long seq);
}

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.

Code changes

1) Message Bus facility

Code Block

public interface MessageBus {

void setMessageSerializer(MessageSerializer messageSerializer);

MessageSerializer getMessageSerializer();


void subscribe(String topic, MessageSubscriber subscriber);

void unsubscribe(String topic, MessageSubscriber subscriber);

void clearAll();

void prune();


void publish(String senderAddress, String topic, PublishScope scope, Object args);

}

MessageBus defines the interface of the message bus facility, it implements a simple publish/subscribe pattern, publishers and subscribers can 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.

MessageBusBase

A simple message bus implementation

MessageHandler

Java annotation for subscriber to specify a message handler

MessageDispatcher

For message subscriber to use to dispatch received messages to annotated message handlers

MessageDetector

To detect interested messages on message bus

2) Job facility

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

3) VM Power state sync

VirtualMachinePowerStateSync

VirtualMachinePowerStateSyncImpl

4) Other refactored classes

VirtualMachineManagerImpl

HighAvailabilityManagerImpl

VirtualMachineGuru

ReservationContext

Hypervisor resource classes

etc.

Schema changes

Code Block

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,

Code Block

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 `parent_id` bigint;

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`.`vm_instance` ADD COLUMN `power_state` VARCHAR(74) 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_vmasync_job_workjoin_jobmap__instancejob_id` FOREIGN KEY (`vm_instance`job_id`) REFERENCES `vm`async_instance`job`(`id`) ON DELETE CASCADE,

  CONSTRAINT INDEX `i_vm_work`fk_async_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 CASCADEjoin_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;

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)

Space shortcuts

Child pages

Versions Compared

Old Version 8

New Version 9

Key

Code changes

2) Job facility

3) VM Power state sync

4) Other refactored classes

Schema changes

Code changes

1) Message Bus facility

2) Job facility

3) VM Power state sync

4) Other refactored classes

Schema changes

API/UI

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 8

New Version 9

Key

Code changes

2) Job facility

3) VM Power state sync

4) Other refactored classes

Schema changes

Code changes

1) Message Bus facility

2) Job facility

3) VM Power state sync

4) Other refactored classes

Schema changes

API/UI