CLOUDSTACK-3272 Latest changes to add global configuration parameters to control the publishing of events on the message bus

engine/components-api/src/com/cloud/event/UsageEventUtils.java

@@ -28,6 +28,7 @@ import javax.inject.Inject;
 import org.apache.log4j.Logger;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.Event;
 import org.apache.cloudstack.framework.events.EventBus;
 import org.apache.cloudstack.framework.events.EventBusException;
@@ -46,6 +47,7 @@ public class UsageEventUtils {
     private static DataCenterDao s_dcDao;
     private static final Logger s_logger = Logger.getLogger(UsageEventUtils.class);
     protected static EventBus s_eventBus = null;
+    protected static ConfigurationDao s_configDao;
 
     @Inject
     UsageEventDao usageEventDao;
@@ -53,6 +55,8 @@ public class UsageEventUtils {
     AccountDao accountDao;
     @Inject
     DataCenterDao dcDao;
+    @Inject
+    ConfigurationDao configDao;
 
     public UsageEventUtils() {
     }
@@ -62,6 +66,7 @@ public class UsageEventUtils {
         s_usageEventDao = usageEventDao;
         s_accountDao = accountDao;
         s_dcDao = dcDao;
+        s_configDao = configDao;
     }
 
     public static void publishUsageEvent(String usageType, long accountId, long zoneId, long resourceId, String resourceName, Long offeringId, Long templateId,
@@ -161,7 +166,11 @@ public class UsageEventUtils {
     }
 
     private static void publishUsageEvent(String usageEventType, Long accountId, Long zoneId, String resourceType, String resourceUUID) {
-
+        String configKey = "publish.usage.events";
+        String value = s_configDao.getValue(configKey);
+        boolean configValue = Boolean.parseBoolean(value);
+        if( !configValue)
+            return;
         try {
             s_eventBus = ComponentContext.getComponent(EventBus.class);
         } catch (NoSuchBeanDefinitionException nbe) {

engine/components-api/src/com/cloud/network/NetworkStateListener.java

@@ -27,6 +27,7 @@ import javax.inject.Inject;
 import org.apache.log4j.Logger;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.EventBus;
 import org.apache.cloudstack.framework.events.EventBusException;
 
@@ -44,14 +45,17 @@ public class NetworkStateListener implements StateListener<State, Event, Network
     protected UsageEventDao _usageEventDao;
     @Inject
     protected NetworkDao _networkDao;
+    @Inject
+    protected ConfigurationDao _configDao;
 
     protected static EventBus s_eventBus = null;
 
     private static final Logger s_logger = Logger.getLogger(NetworkStateListener.class);
 
-    public NetworkStateListener(UsageEventDao usageEventDao, NetworkDao networkDao) {
+    public NetworkStateListener(UsageEventDao usageEventDao, NetworkDao networkDao, ConfigurationDao configDao) {
         _usageEventDao = usageEventDao;
         _networkDao = networkDao;
+        _configDao = configDao;
     }
 
     @Override
@@ -68,6 +72,11 @@ public class NetworkStateListener implements StateListener<State, Event, Network
 
     private void pubishOnEventBus(String event, String status, Network vo, State oldState, State newState) {
 
+        String configKey = "publish.resource.state.events";
+        String value = _configDao.getValue(configKey);
+        boolean configValue = Boolean.parseBoolean(value);
+        if(!configValue)
+            return;
         try {
             s_eventBus = ComponentContext.getComponent(EventBus.class);
         } catch (NoSuchBeanDefinitionException nbe) {

engine/orchestration/src/org/apache/cloudstack/engine/orchestration/NetworkOrchestrator.java

@@ -574,7 +574,7 @@ public class NetworkOrchestrator extends ManagerBase implements NetworkOrchestra
 
         _agentMgr.registerForHostEvents(this, true, false, true);
 
-        Network.State.getStateMachine().registerListener(new NetworkStateListener(_usageEventDao, _networksDao));
+        Network.State.getStateMachine().registerListener(new NetworkStateListener(_usageEventDao, _networksDao, _configDao));
 
         s_logger.info("Network Manager is configured.");
 

server/src/com/cloud/configuration/Config.java

@@ -1915,7 +1915,12 @@ public enum Config {
             "timeout in seconds for each Virtual Router command being aggregated. The final aggregation command timeout would be determined by this timeout * commands counts ",
             null),
 
-    ManagementServerVendor("Advanced", ManagementServer.class, String.class, "mgt.server.vendor", "ACS", "the vendor of management server", null);
+    ManagementServerVendor("Advanced", ManagementServer.class, String.class, "mgt.server.vendor", "ACS", "the vendor of management server", null),
+    PublishActionEvent("Advanced", ManagementServer.class, Boolean.class, "publish.action.events", "true", "enable or disable publishing of action events on the event bus", null),
+    PublishAlertEvent("Advanced", ManagementServer.class, Boolean.class, "publish.alert.events", "true", "enable or disable publishing of alert events on the event bus", null),
+    PublishResourceStateEvent("Advanced", ManagementServer.class, Boolean.class, "publish.resource.state.events", "true", "enable or disable publishing of alert events on the event bus", null),
+    PublishUsageEvent("Advanced", ManagementServer.class, Boolean.class, "publish.usage.events", "true", "enable or disable publishing of usage events on the event bus", null),
+    PublishAsynJobEvent("Advanced", ManagementServer.class, Boolean.class, "publish.async.job.events", "true", "enable or disable publishing of usage events on the event bus", null);
 
     private final String _category;
     private final Class<?> _componentClass;

server/src/com/cloud/event/ActionEventUtils.java

@@ -32,9 +32,11 @@ import org.apache.log4j.Logger;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 
 import org.apache.cloudstack.context.CallContext;
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.EventBus;
 import org.apache.cloudstack.framework.events.EventBusException;
 
+import com.cloud.configuration.Config;
 import com.cloud.domain.Domain;
 import com.cloud.event.dao.EventDao;
 import com.cloud.server.ManagementService;
@@ -56,6 +58,7 @@ public class ActionEventUtils {
     protected static UserDao s_userDao;
     protected static EventBus s_eventBus = null;
     protected static EntityManager s_entityMgr;
+    protected static ConfigurationDao s_configDao;
 
     public static final String EventDetails = "event_details";
     public static final String EventId = "event_id";
@@ -73,6 +76,8 @@ public class ActionEventUtils {
     ProjectDao projectDao;
     @Inject
     EntityManager entityMgr;
+    @Inject
+    ConfigurationDao configDao;
 
     public ActionEventUtils() {
     }
@@ -84,6 +89,7 @@ public class ActionEventUtils {
         s_userDao = userDao;
         s_projectDao = projectDao;
         s_entityMgr = entityMgr;
+        s_configDao = configDao;
     }
 
     public static Long onActionEvent(Long userId, Long accountId, Long domainId, String type, String description) {
@@ -183,6 +189,11 @@ public class ActionEventUtils {
     }
 
     private static void publishOnEventBus(long userId, long accountId, String eventCategory, String eventType, Event.State state, String description) {
+        String configKey = Config.PublishActionEvent.key();
+        String value = s_configDao.getValue(configKey);
+        boolean configValue = Boolean.parseBoolean(value);
+        if(!configValue)
+            return;
         try {
             s_eventBus = ComponentContext.getComponent(EventBus.class);
         } catch (NoSuchBeanDefinitionException nbe) {

server/src/com/cloud/event/AlertGenerator.java

@@ -29,9 +29,11 @@ import org.apache.log4j.Logger;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.stereotype.Component;
 
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.EventBus;
 import org.apache.cloudstack.framework.events.EventBusException;
 
+import com.cloud.configuration.Config;
 import com.cloud.dc.DataCenterVO;
 import com.cloud.dc.HostPodVO;
 import com.cloud.dc.dao.DataCenterDao;
@@ -46,11 +48,14 @@ public class AlertGenerator {
     private static DataCenterDao s_dcDao;
     private static HostPodDao s_podDao;
     protected static EventBus s_eventBus = null;
+    protected static ConfigurationDao s_configDao;
 
     @Inject
     DataCenterDao dcDao;
     @Inject
     HostPodDao podDao;
+    @Inject
+    ConfigurationDao configDao;
 
     public AlertGenerator() {
     }
@@ -62,6 +67,12 @@ public class AlertGenerator {
     }
 
     public static void publishAlertOnEventBus(String alertType, long dataCenterId, Long podId, String subject, String body) {
+
+        String configKey = Config.PublishAlertEvent.key();
+        String value = s_configDao.getValue(configKey);
+        boolean configValue = Boolean.parseBoolean(value);
+        if(!configValue)
+            return;
         try {
             s_eventBus = ComponentContext.getComponent(EventBus.class);
         } catch (NoSuchBeanDefinitionException nbe) {

server/src/com/cloud/network/IpAddressManagerImpl.java

@@ -398,7 +398,7 @@ public class IpAddressManagerImpl extends ManagerBase implements IpAddressManage
         AssignIpAddressFromPodVlanSearch.join("vlan", podVlanSearch, podVlanSearch.entity().getId(), AssignIpAddressFromPodVlanSearch.entity().getVlanId(), JoinType.INNER);
         AssignIpAddressFromPodVlanSearch.done();
 
-        Network.State.getStateMachine().registerListener(new NetworkStateListener(_usageEventDao, _networksDao));
+        Network.State.getStateMachine().registerListener(new NetworkStateListener(_usageEventDao, _networksDao, _configDao));
 
         s_logger.info("Network Manager is configured.");
 

server/src/com/cloud/storage/StorageManagerImpl.java

@@ -488,7 +488,7 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
         LocalStorageSearch.and("type", LocalStorageSearch.entity().getPoolType(), SearchCriteria.Op.IN);
         LocalStorageSearch.done();
 
-        Volume.State.getStateMachine().registerListener(new VolumeStateListener());
+        Volume.State.getStateMachine().registerListener(new VolumeStateListener(_configDao));
 
         return true;
     }

server/src/com/cloud/storage/listener/SnapshotStateListener.java

@@ -22,12 +22,16 @@ import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
 
+import javax.inject.Inject;
+
 import org.apache.log4j.Logger;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.EventBus;
 import org.apache.cloudstack.framework.events.EventBusException;
 
+import com.cloud.configuration.Config;
 import com.cloud.event.EventCategory;
 import com.cloud.server.ManagementService;
 import com.cloud.storage.Snapshot;
@@ -40,6 +44,8 @@ import com.cloud.utils.fsm.StateListener;
 public class SnapshotStateListener implements StateListener<State, Event, SnapshotVO> {
 
     protected static EventBus s_eventBus = null;
+    @Inject
+    private ConfigurationDao _configDao;
 
     private static final Logger s_logger = Logger.getLogger(VolumeStateListener.class);
 
@@ -61,6 +67,11 @@ public class SnapshotStateListener implements StateListener<State, Event, Snapsh
 
     private void pubishOnEventBus(String event, String status, Snapshot vo, State oldState, State newState) {
 
+        String configKey = Config.PublishResourceStateEvent.key();
+        String value = _configDao.getValue(configKey);
+        boolean configValue = Boolean.parseBoolean(value);
+        if(!configValue)
+            return;
         try {
             s_eventBus = ComponentContext.getComponent(EventBus.class);
         } catch (NoSuchBeanDefinitionException nbe) {

server/src/com/cloud/storage/listener/VolumeStateListener.java

@@ -25,9 +25,11 @@ import java.util.Map;
 import org.apache.log4j.Logger;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.EventBus;
 import org.apache.cloudstack.framework.events.EventBusException;
 
+import com.cloud.configuration.Config;
 import com.cloud.event.EventCategory;
 import com.cloud.server.ManagementService;
 import com.cloud.storage.Volume;
@@ -39,11 +41,12 @@ import com.cloud.utils.fsm.StateListener;
 public class VolumeStateListener implements StateListener<State, Event, Volume> {
 
     protected static EventBus s_eventBus = null;
+    protected ConfigurationDao _configDao;
 
     private static final Logger s_logger = Logger.getLogger(VolumeStateListener.class);
 
-    public VolumeStateListener() {
+    public VolumeStateListener(ConfigurationDao configDao) {
-
+        this._configDao = configDao;
     }
 
     @Override
@@ -60,6 +63,11 @@ public class VolumeStateListener implements StateListener<State, Event, Volume>
 
     private void pubishOnEventBus(String event, String status, Volume vo, State oldState, State newState) {
 
+        String configKey = Config.PublishResourceStateEvent.key();
+        String value = _configDao.getValue(configKey);
+        boolean configValue = Boolean.parseBoolean(value);
+        if(!configValue)
+            return;
         try {
             s_eventBus = ComponentContext.getComponent(EventBus.class);
         } catch (NoSuchBeanDefinitionException nbe) {

server/src/com/cloud/test/DatabaseConfig.java

@@ -234,6 +234,11 @@ public class DatabaseConfig {
         s_configurationDescriptions.put("snapshot.test.weeks.per.month", "Set it to a smaller value to take more recurring snapshots");
         s_configurationDescriptions.put("snapshot.test.months.per.year", "Set it to a smaller value to take more recurring snapshots");
         s_configurationDescriptions.put("hypervisor.type", "The type of hypervisor that this deployment will use.");
+        s_configurationDescriptions.put("publish.action.events", "enable or disable to control the publishing of action events on the event bus");
+        s_configurationDescriptions.put("publish.alert.events", "enable or disable to control the publishing of alert events on the event bus");
+        s_configurationDescriptions.put("publish.resource.state.events", "enable or disable to control the publishing of resource state events on the event bus");
+        s_configurationDescriptions.put("publish.usage.events", "enable or disable to control the publishing of usage events on the event bus");
+        s_configurationDescriptions.put("publish.async.job.events", "enable or disable to control the publishing of async job events on the event bus");
 
         s_configurationComponents.put("host.stats.interval", "management-server");
         s_configurationComponents.put("storage.stats.interval", "management-server");
@@ -306,6 +311,11 @@ public class DatabaseConfig {
         s_configurationComponents.put("snapshot.test.weeks.per.month", "SnapshotManager");
         s_configurationComponents.put("snapshot.test.months.per.year", "SnapshotManager");
         s_configurationComponents.put("hypervisor.type", "ManagementServer");
+        s_configurationComponents.put("publish.action.events", "management-server");
+        s_configurationComponents.put("publish.alert.events", "management-server");
+        s_configurationComponents.put("publish.resource.state.events", "management-server");
+        s_configurationComponents.put("publish.usage.events", "management-server");
+        s_configurationComponents.put("publish.async.job.events", "management-server");
 
         s_defaultConfigurationValues.put("host.stats.interval", "60000");
         s_defaultConfigurationValues.put("storage.stats.interval", "60000");
@@ -347,6 +357,11 @@ public class DatabaseConfig {
         s_defaultConfigurationValues.put("init", "false");
         s_defaultConfigurationValues.put("cpu.overprovisioning.factor", "1");
         s_defaultConfigurationValues.put("mem.overprovisioning.factor", "1");
+        s_defaultConfigurationValues.put("publish.action.events", "true");
+        s_defaultConfigurationValues.put("publish.alert.events", "true");
+        s_defaultConfigurationValues.put("publish.resource.state.events", "true");
+        s_defaultConfigurationValues.put("publish.usage.events", "true");
+        s_defaultConfigurationValues.put("publish.async.job.events", "true");
     }
 
     protected DatabaseConfig() {

server/src/com/cloud/vm/UserVmManagerImpl.java

@@ -1611,7 +1611,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
 
         _itMgr.registerGuru(VirtualMachine.Type.User, this);
 
-        VirtualMachine.State.getStateMachine().registerListener(new UserVmStateListener(_usageEventDao, _networkDao, _nicDao, _offeringDao, _vmDao, this));
+        VirtualMachine.State.getStateMachine().registerListener(new UserVmStateListener(_usageEventDao, _networkDao, _nicDao, _offeringDao, _vmDao, this, _configDao));
 
         String value = _configDao.getValue(Config.SetVmInternalNameUsingDisplayName.key());
         _instanceNameFlag = (value == null) ? false : Boolean.parseBoolean(value);

server/src/com/cloud/vm/UserVmStateListener.java

@@ -29,8 +29,10 @@ import com.cloud.vm.dao.UserVmDao;
 import org.apache.log4j.Logger;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.EventBus;
 
+import com.cloud.configuration.Config;
 import com.cloud.event.EventCategory;
 import com.cloud.event.EventTypes;
 import com.cloud.event.UsageEventUtils;
@@ -52,17 +54,20 @@ public class UserVmStateListener implements StateListener<State, VirtualMachine.
     @Inject protected ServiceOfferingDao _offeringDao;
     @Inject protected UserVmDao _userVmDao;
     @Inject protected UserVmManager _userVmMgr;
+    @Inject protected ConfigurationDao _configDao;
     private static final Logger s_logger = Logger.getLogger(UserVmStateListener.class);
 
     protected static EventBus s_eventBus = null;
 
-    public UserVmStateListener(UsageEventDao usageEventDao, NetworkDao networkDao, NicDao nicDao, ServiceOfferingDao offeringDao, UserVmDao userVmDao, UserVmManager userVmMgr) {
+    public UserVmStateListener(UsageEventDao usageEventDao, NetworkDao networkDao, NicDao nicDao, ServiceOfferingDao offeringDao, UserVmDao userVmDao, UserVmManager userVmMgr,
+            ConfigurationDao configDao) {
         this._usageEventDao = usageEventDao;
         this._networkDao = networkDao;
         this._nicDao = nicDao;
         this._offeringDao = offeringDao;
         this._userVmDao = userVmDao;
         this._userVmMgr = userVmMgr;
+        this._configDao = configDao;
     }
 
     @Override
@@ -113,6 +118,11 @@ public class UserVmStateListener implements StateListener<State, VirtualMachine.
 
     private void pubishOnEventBus(String event, String status, VirtualMachine vo, VirtualMachine.State oldState, VirtualMachine.State newState) {
 
+        String configKey = Config.PublishResourceStateEvent.key();
+        String value = _configDao.getValue(configKey);
+        boolean configValue = Boolean.parseBoolean(value);
+        if(!configValue)
+            return;
         try {
             s_eventBus = ComponentContext.getComponent(EventBus.class);
         } catch (NoSuchBeanDefinitionException nbe) {

server/test/org/apache/cloudstack/affinity/AffinityApiUnitTest.java

@@ -57,6 +57,7 @@ import org.apache.cloudstack.affinity.dao.AffinityGroupDao;
 import org.apache.cloudstack.affinity.dao.AffinityGroupDomainMapDao;
 import org.apache.cloudstack.affinity.dao.AffinityGroupVMMapDao;
 import org.apache.cloudstack.context.CallContext;
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.messagebus.MessageBus;
 import org.apache.cloudstack.test.utils.SpringUtils;
 
@@ -298,6 +299,11 @@ public class AffinityApiUnitTest {
             return Mockito.mock(MessageBus.class);
         }
 
+        @Bean
+        public ConfigurationDao configDao() {
+            return Mockito.mock(ConfigurationDao.class);
+        }
+
         public static class Library implements TypeFilter {
 
             @Override