You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stratos.apache.org by im...@apache.org on 2014/04/17 21:25:59 UTC
[3/4] Updated TopologyEventReceiver, TenantEventReceiver,
InstanceNotifierEventReceiver,
HealthStatEventReceiver and implemented separate message queues for each
receiver instance
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/bccad5be/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/AutoscalerTopologyEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/AutoscalerTopologyEventReceiver.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/AutoscalerTopologyEventReceiver.java
new file mode 100644
index 0000000..cf6223c
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/AutoscalerTopologyEventReceiver.java
@@ -0,0 +1,494 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.autoscaler.message.receiver.topology;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.autoscaler.*;
+import org.apache.stratos.autoscaler.deployment.policy.DeploymentPolicy;
+import org.apache.stratos.autoscaler.exception.PartitionValidationException;
+import org.apache.stratos.autoscaler.exception.PolicyValidationException;
+import org.apache.stratos.autoscaler.monitor.AbstractMonitor;
+import org.apache.stratos.autoscaler.monitor.ClusterMonitor;
+import org.apache.stratos.autoscaler.monitor.LbClusterMonitor;
+import org.apache.stratos.autoscaler.partition.PartitionManager;
+import org.apache.stratos.autoscaler.policy.PolicyManager;
+import org.apache.stratos.autoscaler.rule.AutoscalerRuleEvaluator;
+import org.apache.stratos.autoscaler.util.AutoscalerUtil;
+import org.apache.stratos.messaging.domain.topology.Cluster;
+import org.apache.stratos.messaging.domain.topology.Service;
+import org.apache.stratos.messaging.event.Event;
+import org.apache.stratos.messaging.event.topology.*;
+import org.apache.stratos.messaging.listener.topology.*;
+import org.apache.stratos.messaging.message.receiver.topology.TopologyEventReceiver;
+import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
+import org.drools.runtime.StatefulKnowledgeSession;
+import org.drools.runtime.rule.FactHandle;
+
+import java.util.List;
+
+/**
+ * Autoscaler topology receiver.
+ */
+public class AutoscalerTopologyEventReceiver implements Runnable {
+
+ private static final Log log = LogFactory.getLog(AutoscalerTopologyEventReceiver.class);
+
+ private TopologyEventReceiver topologyEventReceiver;
+ private boolean terminated;
+
+ public AutoscalerTopologyEventReceiver() {
+ this.topologyEventReceiver = new TopologyEventReceiver();
+ addEventListeners();
+ }
+
+ @Override
+ public void run() {
+ //FIXME this activated before autoscaler deployer activated.
+ try {
+ Thread.sleep(15000);
+ } catch (InterruptedException ignore) {
+ }
+ Thread thread = new Thread(topologyEventReceiver);
+ thread.start();
+ if (log.isInfoEnabled()) {
+ log.info("Autoscaler topology receiver thread started");
+ }
+
+ // Keep the thread live until terminated
+ while (!terminated) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException ignore) {
+ }
+ }
+ if (log.isInfoEnabled()) {
+ log.info("Autoscaler topology receiver thread terminated");
+ }
+ }
+
+ private void addEventListeners() {
+ // Listen to topology events that affect clusters
+ topologyEventReceiver.addEventListener(new CompleteTopologyEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ try {
+ TopologyManager.acquireReadLock();
+ for (Service service : TopologyManager.getTopology().getServices()) {
+ for (Cluster cluster : service.getClusters()) {
+ startClusterMonitor(cluster);
+ }
+ }
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+
+
+ });
+
+ topologyEventReceiver.addEventListener(new ClusterCreatedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+ try {
+ log.info("Event received: " + event);
+ ClusterCreatedEvent e = (ClusterCreatedEvent) event;
+ TopologyManager.acquireReadLock();
+ Service service = TopologyManager.getTopology().getService(e.getServiceName());
+ Cluster cluster = service.getCluster(e.getClusterId());
+ startClusterMonitor(cluster);
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+
+ });
+
+ topologyEventReceiver.addEventListener(new ClusterRemovedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+ try {
+ ClusterRemovedEvent e = (ClusterRemovedEvent) event;
+ TopologyManager.acquireReadLock();
+
+ String serviceName = e.getServiceName();
+ String clusterId = e.getClusterId();
+ String deploymentPolicy = e.getDeploymentPolicy();
+
+ AbstractMonitor monitor;
+
+ if (e.isLbCluster()) {
+ DeploymentPolicy depPolicy = PolicyManager.getInstance().getDeploymentPolicy(deploymentPolicy);
+ if (depPolicy != null) {
+ List<NetworkPartitionLbHolder> lbHolders = PartitionManager.getInstance()
+ .getNetworkPartitionLbHolders(depPolicy);
+
+ for (NetworkPartitionLbHolder networkPartitionLbHolder : lbHolders) {
+ // removes lb cluster ids
+ boolean isRemoved = networkPartitionLbHolder.removeLbClusterId(clusterId);
+ if (isRemoved) {
+ log.info("Removed the lb cluster [id]:"
+ + clusterId
+ + " reference from Network Partition [id]: "
+ + networkPartitionLbHolder
+ .getNetworkPartitionId());
+
+ }
+ if (log.isDebugEnabled()) {
+ log.debug(networkPartitionLbHolder);
+ }
+
+ }
+ }
+ monitor = AutoscalerContext.getInstance()
+ .removeLbMonitor(clusterId);
+
+ } else {
+ monitor = AutoscalerContext.getInstance()
+ .removeMonitor(clusterId);
+ }
+
+ // runTerminateAllRule(monitor);
+ if (monitor != null) {
+ monitor.destroy();
+ log.info(String.format("Cluster monitor has been removed successfully: [cluster] %s ",
+ clusterId));
+ }
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+
+ });
+
+ topologyEventReceiver.addEventListener(new MemberStartedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ }
+
+ });
+
+ topologyEventReceiver.addEventListener(new MemberTerminatedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ try {
+ TopologyManager.acquireReadLock();
+ MemberTerminatedEvent e = (MemberTerminatedEvent) event;
+ String networkPartitionId = e.getNetworkPartitionId();
+ String clusterId = e.getClusterId();
+ String partitionId = e.getPartitionId();
+ AbstractMonitor monitor;
+
+ if (AutoscalerContext.getInstance().monitorExist(clusterId)) {
+ monitor = AutoscalerContext.getInstance().getMonitor(clusterId);
+ } else {
+ //This is LB member
+ monitor = AutoscalerContext.getInstance().getLBMonitor(clusterId);
+ }
+
+ NetworkPartitionContext networkPartitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId);
+
+ PartitionContext partitionContext = networkPartitionContext.getPartitionCtxt(partitionId);
+ String memberId = e.getMemberId();
+ partitionContext.removeMemberStatsContext(memberId);
+
+ if (partitionContext.removeTerminationPendingMember(memberId)) {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Member is removed from termination pending members list: [member] %s", memberId));
+ }
+ } else if (partitionContext.removePendingMember(memberId)) {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Member is removed from pending members list: [member] %s", memberId));
+ }
+ } else if (partitionContext.removeActiveMemberById(memberId)) {
+ log.warn(String.format("Member is in the wrong list and it is removed from active members list", memberId));
+ } else {
+ log.warn(String.format("Member is not available in any of the list active, pending and termination pending", memberId));
+ }
+
+ if (log.isInfoEnabled()) {
+ log.info(String.format("Member stat context has been removed successfully: [member] %s", memberId));
+ }
+// partitionContext.decrementCurrentActiveMemberCount(1);
+
+
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+
+ });
+
+ topologyEventReceiver.addEventListener(new MemberActivatedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ try {
+ TopologyManager.acquireReadLock();
+
+ MemberActivatedEvent e = (MemberActivatedEvent) event;
+ String memberId = e.getMemberId();
+ String partitionId = e.getPartitionId();
+ String networkPartitionId = e.getNetworkPartitionId();
+
+ PartitionContext partitionContext;
+ String clusterId = e.getClusterId();
+ AbstractMonitor monitor;
+
+ if (AutoscalerContext.getInstance().monitorExist(clusterId)) {
+ monitor = AutoscalerContext.getInstance().getMonitor(clusterId);
+ partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
+ } else {
+ monitor = AutoscalerContext.getInstance().getLBMonitor(clusterId);
+ partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
+ }
+ partitionContext.addMemberStatsContext(new MemberStatsContext(memberId));
+ if (log.isInfoEnabled()) {
+ log.info(String.format("Member stat context has been added successfully: [member] %s", memberId));
+ }
+// partitionContext.incrementCurrentActiveMemberCount(1);
+ partitionContext.movePendingMemberToActiveMembers(memberId);
+
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+ });
+
+
+ topologyEventReceiver.addEventListener(new MemberMaintenanceListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ try {
+ TopologyManager.acquireReadLock();
+
+ MemberMaintenanceModeEvent e = (MemberMaintenanceModeEvent) event;
+ String memberId = e.getMemberId();
+ String partitionId = e.getPartitionId();
+ String networkPartitionId = e.getNetworkPartitionId();
+
+ PartitionContext partitionContext;
+ String clusterId = e.getClusterId();
+ AbstractMonitor monitor;
+
+ if (AutoscalerContext.getInstance().monitorExist(clusterId)) {
+ monitor = AutoscalerContext.getInstance().getMonitor(clusterId);
+ partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
+ } else {
+ monitor = AutoscalerContext.getInstance().getLBMonitor(clusterId);
+ partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
+ }
+ partitionContext.addMemberStatsContext(new MemberStatsContext(memberId));
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Member has been moved as pending termination: [member] %s", memberId));
+ }
+ partitionContext.moveActiveMemberToTerminationPendingMembers(memberId);
+
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+ });
+
+ topologyEventReceiver.addEventListener(new ServiceRemovedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+// try {
+// TopologyManager.acquireReadLock();
+//
+// // Remove all clusters of given service from context
+// ServiceRemovedEvent serviceRemovedEvent = (ServiceRemovedEvent)event;
+// for(Service service : TopologyManager.getTopology().getServices()) {
+// for(Cluster cluster : service.getClusters()) {
+// removeMonitor(cluster.getHostName());
+// }
+// }
+// }
+// finally {
+// TopologyManager.releaseReadLock();
+// }
+ }
+ });
+ }
+
+ private class LBClusterMonitorAdder implements Runnable {
+ private Cluster cluster;
+
+ public LBClusterMonitorAdder(Cluster cluster) {
+ this.cluster = cluster;
+ }
+
+ public void run() {
+ LbClusterMonitor monitor = null;
+ int retries = 5;
+ boolean success = false;
+ do {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e1) {
+ }
+ try {
+ monitor = AutoscalerUtil.getLBClusterMonitor(cluster);
+ success = true;
+
+ } catch (PolicyValidationException e) {
+ String msg = "LB Cluster monitor creation failed for cluster: " + cluster.getClusterId();
+ log.debug(msg, e);
+ retries--;
+
+ } catch (PartitionValidationException e) {
+ String msg = "LB Cluster monitor creation failed for cluster: " + cluster.getClusterId();
+ log.debug(msg, e);
+ retries--;
+ }
+ } while (!success && retries <= 0);
+
+ if (monitor == null) {
+ String msg = "LB Cluster monitor creation failed, even after retrying for 5 times, "
+ + "for cluster: " + cluster.getClusterId();
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ Thread th = new Thread(monitor);
+ th.start();
+ AutoscalerContext.getInstance().addLbMonitor(monitor);
+ if (log.isInfoEnabled()) {
+ log.info(String.format("LB Cluster monitor has been added successfully: [cluster] %s",
+ cluster.getClusterId()));
+ }
+ }
+ }
+
+ private class ClusterMonitorAdder implements Runnable {
+ private Cluster cluster;
+
+ public ClusterMonitorAdder(Cluster cluster) {
+ this.cluster = cluster;
+ }
+
+ public void run() {
+ ClusterMonitor monitor = null;
+ int retries = 5;
+ boolean success = false;
+ do {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e1) {
+ }
+
+ try {
+ monitor = AutoscalerUtil.getClusterMonitor(cluster);
+ success = true;
+
+ } catch (PolicyValidationException e) {
+ String msg = "Cluster monitor creation failed for cluster: " + cluster.getClusterId();
+ log.debug(msg, e);
+ retries--;
+
+ } catch (PartitionValidationException e) {
+ String msg = "Cluster monitor creation failed for cluster: " + cluster.getClusterId();
+ log.debug(msg, e);
+ retries--;
+ }
+ } while (!success && retries != 0);
+
+ if (monitor == null) {
+ String msg = "Cluster monitor creation failed, even after retrying for 5 times, "
+ + "for cluster: " + cluster.getClusterId();
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ Thread th = new Thread(monitor);
+ th.start();
+ AutoscalerContext.getInstance().addMonitor(monitor);
+ if (log.isInfoEnabled()) {
+ log.info(String.format("Cluster monitor has been added successfully: [cluster] %s",
+ cluster.getClusterId()));
+ }
+ }
+ }
+
+ private void runTerminateAllRule(AbstractMonitor monitor) {
+
+ FactHandle terminateAllFactHandle = null;
+
+ StatefulKnowledgeSession terminateAllKnowledgeSession = null;
+
+ for (NetworkPartitionContext networkPartitionContext : monitor.getNetworkPartitionCtxts().values()) {
+ terminateAllFactHandle = AutoscalerRuleEvaluator.evaluateTerminateAll(terminateAllKnowledgeSession
+ , terminateAllFactHandle, networkPartitionContext);
+ }
+
+ }
+
+ /**
+ * Terminate load balancer topology receiver thread.
+ */
+ public void terminate() {
+ topologyEventReceiver.terminate();
+ terminated = true;
+ }
+
+ protected synchronized void startClusterMonitor(Cluster cluster) {
+ Thread th = null;
+ if (cluster.isLbCluster()
+ && !AutoscalerContext.getInstance()
+ .lbMonitorExist(
+ cluster.getClusterId())) {
+ th = new Thread(new LBClusterMonitorAdder(
+ cluster));
+ } else if (!cluster.isLbCluster() && !AutoscalerContext.getInstance()
+ .monitorExist(cluster.getClusterId())) {
+ th = new Thread(
+ new ClusterMonitorAdder(cluster));
+ }
+ if (th != null) {
+ th.start();
+ try {
+ th.join();
+ } catch (InterruptedException ignore) {
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug(String
+ .format("Cluster monitor thread has been started successfully: [cluster] %s ",
+ cluster.getClusterId()));
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/bccad5be/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/AutoscalerTopologyReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/AutoscalerTopologyReceiver.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/AutoscalerTopologyReceiver.java
deleted file mode 100644
index 6443a8a..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/AutoscalerTopologyReceiver.java
+++ /dev/null
@@ -1,502 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.stratos.autoscaler.message.receiver.topology;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.*;
-import org.apache.stratos.autoscaler.deployment.policy.DeploymentPolicy;
-import org.apache.stratos.autoscaler.exception.PartitionValidationException;
-import org.apache.stratos.autoscaler.exception.PolicyValidationException;
-import org.apache.stratos.autoscaler.monitor.AbstractMonitor;
-import org.apache.stratos.autoscaler.monitor.ClusterMonitor;
-import org.apache.stratos.autoscaler.monitor.LbClusterMonitor;
-import org.apache.stratos.autoscaler.partition.PartitionManager;
-import org.apache.stratos.autoscaler.policy.PolicyManager;
-import org.apache.stratos.autoscaler.rule.AutoscalerRuleEvaluator;
-import org.apache.stratos.autoscaler.util.AutoscalerUtil;
-import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.Service;
-import org.apache.stratos.messaging.event.Event;
-import org.apache.stratos.messaging.event.topology.*;
-import org.apache.stratos.messaging.listener.topology.*;
-import org.apache.stratos.messaging.message.processor.topology.TopologyMessageProcessorChain;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyEventMessageDelegator;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyReceiver;
-import org.drools.runtime.StatefulKnowledgeSession;
-import org.drools.runtime.rule.FactHandle;
-
-import java.util.List;
-
-/**
- * Autoscaler topology receiver.
- */
-public class AutoscalerTopologyReceiver implements Runnable {
-
- private static final Log log = LogFactory.getLog(AutoscalerTopologyReceiver.class);
-
- private TopologyReceiver topologyReceiver;
- private boolean terminated;
-
- public AutoscalerTopologyReceiver() {
- this.topologyReceiver = new TopologyReceiver(createMessageDelegator(), new TopologyEventMessageListener());
- }
-
- @Override
- public void run() {
- //FIXME this activated before autoscaler deployer activated.
- try {
- Thread.sleep(15000);
- } catch (InterruptedException ignore) {
- }
- Thread thread = new Thread(topologyReceiver);
- thread.start();
- if (log.isInfoEnabled()) {
- log.info("Autoscaler topology receiver thread started");
- }
-
- // Keep the thread live until terminated
- while (!terminated) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ignore) {
- }
- }
- if (log.isInfoEnabled()) {
- log.info("Autoscaler topology receiver thread terminated");
- }
- }
-
- private TopologyEventMessageDelegator createMessageDelegator() {
- TopologyMessageProcessorChain processorChain = createEventProcessorChain();
- return new TopologyEventMessageDelegator(processorChain, TopologyEventMessageQueue.getInstance());
- }
-
- private TopologyMessageProcessorChain createEventProcessorChain() {
- // Listen to topology events that affect clusters
- TopologyMessageProcessorChain processorChain = new TopologyMessageProcessorChain();
- processorChain.addEventListener(new CompleteTopologyEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- try {
- TopologyManager.acquireReadLock();
- for (Service service : TopologyManager.getTopology().getServices()) {
- for (Cluster cluster : service.getClusters()) {
- startClusterMonitor(cluster);
- }
- }
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
-
- });
-
- processorChain.addEventListener(new ClusterCreatedEventListener() {
- @Override
- protected void onEvent(Event event) {
- try {
- log.info("Event received: " + event);
- ClusterCreatedEvent e = (ClusterCreatedEvent) event;
- TopologyManager.acquireReadLock();
- Service service = TopologyManager.getTopology().getService(e.getServiceName());
- Cluster cluster = service.getCluster(e.getClusterId());
- startClusterMonitor(cluster);
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- });
-
- processorChain.addEventListener(new ClusterRemovedEventListener() {
- @Override
- protected void onEvent(Event event) {
- try {
- ClusterRemovedEvent e = (ClusterRemovedEvent) event;
- TopologyManager.acquireReadLock();
-
- String serviceName = e.getServiceName();
- String clusterId = e.getClusterId();
- String deploymentPolicy = e.getDeploymentPolicy();
-
- AbstractMonitor monitor;
-
- if (e.isLbCluster()) {
- DeploymentPolicy depPolicy = PolicyManager.getInstance().getDeploymentPolicy(deploymentPolicy);
- if (depPolicy != null) {
- List<NetworkPartitionLbHolder> lbHolders = PartitionManager.getInstance()
- .getNetworkPartitionLbHolders(depPolicy);
-
- for (NetworkPartitionLbHolder networkPartitionLbHolder : lbHolders) {
- // removes lb cluster ids
- boolean isRemoved = networkPartitionLbHolder.removeLbClusterId(clusterId);
- if (isRemoved) {
- log.info("Removed the lb cluster [id]:"
- + clusterId
- + " reference from Network Partition [id]: "
- + networkPartitionLbHolder
- .getNetworkPartitionId());
-
- }
- if (log.isDebugEnabled()) {
- log.debug(networkPartitionLbHolder);
- }
-
- }
- }
- monitor = AutoscalerContext.getInstance()
- .removeLbMonitor(clusterId);
-
- } else {
- monitor = AutoscalerContext.getInstance()
- .removeMonitor(clusterId);
- }
-
- // runTerminateAllRule(monitor);
- if (monitor != null) {
- monitor.destroy();
- log.info(String.format("Cluster monitor has been removed successfully: [cluster] %s ",
- clusterId));
- }
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- });
-
- processorChain.addEventListener(new MemberStartedEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- }
-
- });
-
- processorChain.addEventListener(new MemberTerminatedEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- try {
- TopologyManager.acquireReadLock();
- MemberTerminatedEvent e = (MemberTerminatedEvent) event;
- String networkPartitionId = e.getNetworkPartitionId();
- String clusterId = e.getClusterId();
- String partitionId = e.getPartitionId();
- AbstractMonitor monitor;
-
- if (AutoscalerContext.getInstance().monitorExist(clusterId)) {
- monitor = AutoscalerContext.getInstance().getMonitor(clusterId);
- } else {
- //This is LB member
- monitor = AutoscalerContext.getInstance().getLBMonitor(clusterId);
- }
-
- NetworkPartitionContext networkPartitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId);
-
- PartitionContext partitionContext = networkPartitionContext.getPartitionCtxt(partitionId);
- String memberId = e.getMemberId();
- partitionContext.removeMemberStatsContext(memberId);
-
- if (partitionContext.removeTerminationPendingMember(memberId)) {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Member is removed from termination pending members list: [member] %s", memberId));
- }
- } else if (partitionContext.removePendingMember(memberId)) {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Member is removed from pending members list: [member] %s", memberId));
- }
- } else if (partitionContext.removeActiveMemberById(memberId)) {
- log.warn(String.format("Member is in the wrong list and it is removed from active members list", memberId));
- } else {
- log.warn(String.format("Member is not available in any of the list active, pending and termination pending", memberId));
- }
-
- if (log.isInfoEnabled()) {
- log.info(String.format("Member stat context has been removed successfully: [member] %s", memberId));
- }
-// partitionContext.decrementCurrentActiveMemberCount(1);
-
-
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- });
-
- processorChain.addEventListener(new MemberActivatedEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- try {
- TopologyManager.acquireReadLock();
-
- MemberActivatedEvent e = (MemberActivatedEvent) event;
- String memberId = e.getMemberId();
- String partitionId = e.getPartitionId();
- String networkPartitionId = e.getNetworkPartitionId();
-
- PartitionContext partitionContext;
- String clusterId = e.getClusterId();
- AbstractMonitor monitor;
-
- if (AutoscalerContext.getInstance().monitorExist(clusterId)) {
- monitor = AutoscalerContext.getInstance().getMonitor(clusterId);
- partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
- } else {
- monitor = AutoscalerContext.getInstance().getLBMonitor(clusterId);
- partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
- }
- partitionContext.addMemberStatsContext(new MemberStatsContext(memberId));
- if (log.isInfoEnabled()) {
- log.info(String.format("Member stat context has been added successfully: [member] %s", memberId));
- }
-// partitionContext.incrementCurrentActiveMemberCount(1);
- partitionContext.movePendingMemberToActiveMembers(memberId);
-
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
- });
-
-
- processorChain.addEventListener(new MemberMaintenanceListener() {
- @Override
- protected void onEvent(Event event) {
-
- try {
- TopologyManager.acquireReadLock();
-
- MemberMaintenanceModeEvent e = (MemberMaintenanceModeEvent) event;
- String memberId = e.getMemberId();
- String partitionId = e.getPartitionId();
- String networkPartitionId = e.getNetworkPartitionId();
-
- PartitionContext partitionContext;
- String clusterId = e.getClusterId();
- AbstractMonitor monitor;
-
- if (AutoscalerContext.getInstance().monitorExist(clusterId)) {
- monitor = AutoscalerContext.getInstance().getMonitor(clusterId);
- partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
- } else {
- monitor = AutoscalerContext.getInstance().getLBMonitor(clusterId);
- partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
- }
- partitionContext.addMemberStatsContext(new MemberStatsContext(memberId));
- if (log.isDebugEnabled()) {
- log.debug(String.format("Member has been moved as pending termination: [member] %s", memberId));
- }
- partitionContext.moveActiveMemberToTerminationPendingMembers(memberId);
-
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
- });
-
- processorChain.addEventListener(new ServiceRemovedEventListener() {
- @Override
- protected void onEvent(Event event) {
-// try {
-// TopologyManager.acquireReadLock();
-//
-// // Remove all clusters of given service from context
-// ServiceRemovedEvent serviceRemovedEvent = (ServiceRemovedEvent)event;
-// for(Service service : TopologyManager.getTopology().getServices()) {
-// for(Cluster cluster : service.getClusters()) {
-// removeMonitor(cluster.getHostName());
-// }
-// }
-// }
-// finally {
-// TopologyManager.releaseReadLock();
-// }
- }
- });
- return processorChain;
- }
-
- private class LBClusterMonitorAdder implements Runnable {
- private Cluster cluster;
-
- public LBClusterMonitorAdder(Cluster cluster) {
- this.cluster = cluster;
- }
-
- public void run() {
- LbClusterMonitor monitor = null;
- int retries = 5;
- boolean success = false;
- do {
- try {
- Thread.sleep(5000);
- } catch (InterruptedException e1) {
- }
- try {
- monitor = AutoscalerUtil.getLBClusterMonitor(cluster);
- success = true;
-
- } catch (PolicyValidationException e) {
- String msg = "LB Cluster monitor creation failed for cluster: " + cluster.getClusterId();
- log.debug(msg, e);
- retries--;
-
- } catch (PartitionValidationException e) {
- String msg = "LB Cluster monitor creation failed for cluster: " + cluster.getClusterId();
- log.debug(msg, e);
- retries--;
- }
- } while (!success && retries <= 0);
-
- if (monitor == null) {
- String msg = "LB Cluster monitor creation failed, even after retrying for 5 times, "
- + "for cluster: " + cluster.getClusterId();
- log.error(msg);
- throw new RuntimeException(msg);
- }
-
- Thread th = new Thread(monitor);
- th.start();
- AutoscalerContext.getInstance().addLbMonitor(monitor);
- if (log.isInfoEnabled()) {
- log.info(String.format("LB Cluster monitor has been added successfully: [cluster] %s",
- cluster.getClusterId()));
- }
- }
- }
-
- private class ClusterMonitorAdder implements Runnable {
- private Cluster cluster;
-
- public ClusterMonitorAdder(Cluster cluster) {
- this.cluster = cluster;
- }
-
- public void run() {
- ClusterMonitor monitor = null;
- int retries = 5;
- boolean success = false;
- do {
- try {
- Thread.sleep(5000);
- } catch (InterruptedException e1) {
- }
-
- try {
- monitor = AutoscalerUtil.getClusterMonitor(cluster);
- success = true;
-
- } catch (PolicyValidationException e) {
- String msg = "Cluster monitor creation failed for cluster: " + cluster.getClusterId();
- log.debug(msg, e);
- retries--;
-
- } catch (PartitionValidationException e) {
- String msg = "Cluster monitor creation failed for cluster: " + cluster.getClusterId();
- log.debug(msg, e);
- retries--;
- }
- } while (!success && retries != 0);
-
- if (monitor == null) {
- String msg = "Cluster monitor creation failed, even after retrying for 5 times, "
- + "for cluster: " + cluster.getClusterId();
- log.error(msg);
- throw new RuntimeException(msg);
- }
-
- Thread th = new Thread(monitor);
- th.start();
- AutoscalerContext.getInstance().addMonitor(monitor);
- if (log.isInfoEnabled()) {
- log.info(String.format("Cluster monitor has been added successfully: [cluster] %s",
- cluster.getClusterId()));
- }
- }
- }
-
- private void runTerminateAllRule(AbstractMonitor monitor) {
-
- FactHandle terminateAllFactHandle = null;
-
- StatefulKnowledgeSession terminateAllKnowledgeSession = null;
-
- for (NetworkPartitionContext networkPartitionContext : monitor.getNetworkPartitionCtxts().values()) {
- terminateAllFactHandle = AutoscalerRuleEvaluator.evaluateTerminateAll(terminateAllKnowledgeSession
- , terminateAllFactHandle, networkPartitionContext);
- }
-
- }
-
- /**
- * Terminate load balancer topology receiver thread.
- */
- public void terminate() {
- topologyReceiver.terminate();
- terminated = true;
- }
-
- protected synchronized void startClusterMonitor(Cluster cluster) {
- Thread th = null;
- if (cluster.isLbCluster()
- && !AutoscalerContext.getInstance()
- .lbMonitorExist(
- cluster.getClusterId())) {
- th = new Thread(new LBClusterMonitorAdder(
- cluster));
- } else if (!cluster.isLbCluster() && !AutoscalerContext.getInstance()
- .monitorExist(cluster.getClusterId())) {
- th = new Thread(
- new ClusterMonitorAdder(cluster));
- }
- if (th != null) {
- th.start();
- try {
- th.join();
- } catch (InterruptedException ignore) {
- }
-
- if (log.isDebugEnabled()) {
- log.debug(String
- .format("Cluster monitor thread has been started successfully: [cluster] %s ",
- cluster.getClusterId()));
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/bccad5be/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageListener.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageListener.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageListener.java
deleted file mode 100644
index 04685dd..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageListener.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.stratos.autoscaler.message.receiver.topology;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.TextMessage;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-/**
- * Implements functionality for receiving text based event messages from the topology
- * message broker topic and add them to the event queue.
- */
-public class TopologyEventMessageListener implements MessageListener {
-
- private static final Log log = LogFactory.getLog(TopologyEventMessageListener.class);
-
- @Override
- public void onMessage(Message message) {
- if (message instanceof TextMessage) {
- TextMessage receivedMessage = (TextMessage) message;
- try {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Topology message received: %s", ((TextMessage) message).getText()));
- }
- // Add received message to the queue
- TopologyEventMessageQueue.getInstance().add(receivedMessage);
-
- } catch (JMSException e) {
- log.error(e.getMessage(), e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/bccad5be/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageQueue.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageQueue.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageQueue.java
deleted file mode 100644
index aeb08b3..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageQueue.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.stratos.autoscaler.message.receiver.topology;
-
-import java.util.concurrent.LinkedBlockingQueue;
-
-import javax.jms.TextMessage;
-
-/**
- * Implements a blocking queue for managing topology event messages.
- */
-public class TopologyEventMessageQueue extends LinkedBlockingQueue<TextMessage>{
-
- private static class InstanceHolder {
- private static final TopologyEventMessageQueue INSTANCE = new TopologyEventMessageQueue();
- }
-
- public static TopologyEventMessageQueue getInstance() {
- return InstanceHolder.INSTANCE;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/bccad5be/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
index 66c776f..e4ddaed 100644
--- a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
+++ b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
@@ -41,9 +41,7 @@ import org.apache.stratos.messaging.event.instance.notifier.InstanceCleanupMembe
import org.apache.stratos.messaging.listener.instance.notifier.ArtifactUpdateEventListener;
import org.apache.stratos.messaging.listener.instance.notifier.InstanceCleanupClusterEventListener;
import org.apache.stratos.messaging.listener.instance.notifier.InstanceCleanupMemberEventListener;
-import org.apache.stratos.messaging.message.processor.instance.notifier.InstanceNotifierMessageProcessorChain;
-import org.apache.stratos.messaging.message.receiver.instance.notifier.InstanceNotifierEventMessageDelegator;
-import org.apache.stratos.messaging.message.receiver.instance.notifier.InstanceNotifierEventMessageReceiver;
+import org.apache.stratos.messaging.message.receiver.instance.notifier.InstanceNotifierEventReceiver;
import java.util.List;
import java.util.concurrent.Executors;
@@ -126,43 +124,42 @@ public class CartridgeAgent implements Runnable {
if(log.isDebugEnabled()) {
log.debug("Starting instance notifier event message receiver thread");
}
- InstanceNotifierMessageProcessorChain processorChain = new InstanceNotifierMessageProcessorChain();
- processorChain.addEventListener(new ArtifactUpdateEventListener() {
+
+ InstanceNotifierEventReceiver eventReceiver = new InstanceNotifierEventReceiver();
+ eventReceiver.addEventListener(new ArtifactUpdateEventListener() {
@Override
protected void onEvent(Event event) {
onArtifactUpdateEvent((ArtifactUpdatedEvent) event);
}
});
- processorChain.addEventListener(new InstanceCleanupMemberEventListener() {
+ eventReceiver.addEventListener(new InstanceCleanupMemberEventListener() {
@Override
protected void onEvent(Event event) {
String memberIdInPayload = CartridgeAgentConfiguration.getInstance().getMemberId();
- InstanceCleanupMemberEvent instanceCleanupMemberEvent = (InstanceCleanupMemberEvent)event;
- if(memberIdInPayload.equals(instanceCleanupMemberEvent.getMemberId())) {
+ InstanceCleanupMemberEvent instanceCleanupMemberEvent = (InstanceCleanupMemberEvent) event;
+ if (memberIdInPayload.equals(instanceCleanupMemberEvent.getMemberId())) {
onInstanceCleanupEvent();
}
}
});
- processorChain.addEventListener(new InstanceCleanupClusterEventListener() {
+ eventReceiver.addEventListener(new InstanceCleanupClusterEventListener() {
@Override
protected void onEvent(Event event) {
String clusterIdInPayload = CartridgeAgentConfiguration.getInstance().getClusterId();
- InstanceCleanupClusterEvent instanceCleanupClusterEvent = (InstanceCleanupClusterEvent)event;
- if(clusterIdInPayload.equals(instanceCleanupClusterEvent.getClusterId())) {
+ InstanceCleanupClusterEvent instanceCleanupClusterEvent = (InstanceCleanupClusterEvent) event;
+ if (clusterIdInPayload.equals(instanceCleanupClusterEvent.getClusterId())) {
onInstanceCleanupEvent();
}
}
});
- InstanceNotifierEventMessageDelegator messageDelegator = new InstanceNotifierEventMessageDelegator(processorChain);
- InstanceNotifierEventMessageReceiver messageReceiver = new InstanceNotifierEventMessageReceiver(messageDelegator);
- Thread messageReceiverThread = new Thread(messageReceiver);
- messageReceiverThread.start();
+ Thread eventReceiverThread = new Thread(eventReceiver);
+ eventReceiverThread.start();
// Wait until message receiver is subscribed to the topic to
// send the instance started event
- while (!messageReceiver.isSubscribed()) {
+ while (!eventReceiver.isSubscribed()) {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/bccad5be/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java b/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java
index 8b4ba18..44b7419 100644
--- a/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java
+++ b/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java
@@ -25,10 +25,8 @@ import org.apache.stratos.load.balancer.common.statistics.LoadBalancerStatistics
import org.apache.stratos.load.balancer.common.statistics.notifier.LoadBalancerStatisticsNotifier;
import org.apache.stratos.messaging.event.Event;
import org.apache.stratos.messaging.listener.topology.*;
-import org.apache.stratos.messaging.message.processor.topology.TopologyMessageProcessorChain;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyEventMessageDelegator;
+import org.apache.stratos.messaging.message.receiver.topology.TopologyEventReceiver;
import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyReceiver;
/**
* Load balancer extension thread for executing load balancer life-cycle according to the topology updates
@@ -40,7 +38,7 @@ public class LoadBalancerExtension implements Runnable {
private LoadBalancer loadBalancer;
private LoadBalancerStatisticsReader statsReader;
private boolean loadBalancerStarted;
- private TopologyReceiver topologyReceiver;
+ private TopologyEventReceiver topologyEventReceiver;
private LoadBalancerStatisticsNotifier statisticsNotifier;
private boolean terminated;
@@ -62,8 +60,9 @@ public class LoadBalancerExtension implements Runnable {
}
// Start topology receiver thread
- topologyReceiver = new TopologyReceiver(createMessageDelegator());
- Thread topologyReceiverThread = new Thread(topologyReceiver);
+ topologyEventReceiver = new TopologyEventReceiver();
+ addEventListeners();
+ Thread topologyReceiverThread = new Thread(topologyEventReceiver);
topologyReceiverThread.start();
if(statsReader != null) {
@@ -87,9 +86,8 @@ public class LoadBalancerExtension implements Runnable {
}
}
- private TopologyEventMessageDelegator createMessageDelegator() {
- TopologyMessageProcessorChain processorChain = createEventProcessorChain();
- processorChain.addEventListener(new CompleteTopologyEventListener() {
+ private void addEventListeners() {
+ topologyEventReceiver.addEventListener(new CompleteTopologyEventListener() {
@Override
protected void onEvent(Event event) {
@@ -108,42 +106,36 @@ public class LoadBalancerExtension implements Runnable {
}
}
});
- return new TopologyEventMessageDelegator(processorChain);
- }
-
- private TopologyMessageProcessorChain createEventProcessorChain() {
- TopologyMessageProcessorChain processorChain = new TopologyMessageProcessorChain();
- processorChain.addEventListener(new MemberActivatedEventListener() {
+ topologyEventReceiver.addEventListener(new MemberActivatedEventListener() {
@Override
protected void onEvent(Event event) {
reloadConfiguration();
}
});
- processorChain.addEventListener(new MemberSuspendedEventListener() {
+ topologyEventReceiver.addEventListener(new MemberSuspendedEventListener() {
@Override
protected void onEvent(Event event) {
reloadConfiguration();
}
});
- processorChain.addEventListener(new MemberTerminatedEventListener() {
+ topologyEventReceiver.addEventListener(new MemberTerminatedEventListener() {
@Override
protected void onEvent(Event event) {
reloadConfiguration();
}
});
- processorChain.addEventListener(new ClusterRemovedEventListener() {
+ topologyEventReceiver.addEventListener(new ClusterRemovedEventListener() {
@Override
protected void onEvent(Event event) {
reloadConfiguration();
}
});
- processorChain.addEventListener(new ServiceRemovedEventListener() {
+ topologyEventReceiver.addEventListener(new ServiceRemovedEventListener() {
@Override
protected void onEvent(Event event) {
reloadConfiguration();
}
});
- return processorChain;
}
private void reloadConfiguration() {
@@ -159,8 +151,8 @@ public class LoadBalancerExtension implements Runnable {
}
public void terminate() {
- if(topologyReceiver != null) {
- topologyReceiver.terminate();
+ if(topologyEventReceiver != null) {
+ topologyEventReceiver.terminate();
}
if(statisticsNotifier != null) {
statisticsNotifier.terminate();
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/bccad5be/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerTenantEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerTenantEventReceiver.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerTenantEventReceiver.java
new file mode 100644
index 0000000..ad1907a
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerTenantEventReceiver.java
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.load.balancer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.load.balancer.context.LoadBalancerContext;
+import org.apache.stratos.messaging.domain.tenant.Tenant;
+import org.apache.stratos.messaging.domain.topology.Cluster;
+import org.apache.stratos.messaging.domain.topology.Service;
+import org.apache.stratos.messaging.domain.topology.ServiceType;
+import org.apache.stratos.messaging.event.Event;
+import org.apache.stratos.messaging.event.tenant.CompleteTenantEvent;
+import org.apache.stratos.messaging.event.tenant.TenantSubscribedEvent;
+import org.apache.stratos.messaging.event.tenant.TenantUnSubscribedEvent;
+import org.apache.stratos.messaging.listener.tenant.CompleteTenantEventListener;
+import org.apache.stratos.messaging.listener.tenant.TenantSubscribedEventListener;
+import org.apache.stratos.messaging.listener.tenant.TenantUnSubscribedEventListener;
+import org.apache.stratos.messaging.message.receiver.tenant.TenantEventReceiver;
+import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Load balancer tenant receiver updates load balancer context according to
+ * incoming tenant events.
+ */
+public class LoadBalancerTenantEventReceiver implements Runnable {
+
+ private static final Log log = LogFactory.getLog(LoadBalancerTenantEventReceiver.class);
+
+ private final TenantEventReceiver tenantEventReceiver;
+ private boolean terminated;
+
+ public LoadBalancerTenantEventReceiver() {
+ tenantEventReceiver = new TenantEventReceiver();
+ addEventListeners();
+ }
+
+ private void addEventListeners() {
+ tenantEventReceiver.addEventListener(new CompleteTenantEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+ CompleteTenantEvent completeTenantEvent = (CompleteTenantEvent) event;
+ for (Tenant tenant : completeTenantEvent.getTenants()) {
+ for (String serviceName : tenant.getServiceSubscriptions()) {
+ if(isMultiTenantService(serviceName)) {
+ addTenantSubscriptionToLbContext(serviceName, tenant.getTenantId());
+ }
+ }
+ }
+ }
+ });
+ tenantEventReceiver.addEventListener(new TenantSubscribedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+ TenantSubscribedEvent tenantSubscribedEvent = (TenantSubscribedEvent) event;
+ if(log.isDebugEnabled()) {
+ log.debug(String.format("Tenant subscribed event received: [tenant-id] %d [service] %s",
+ tenantSubscribedEvent.getTenantId(), tenantSubscribedEvent.getServiceName()));
+ }
+ if(isMultiTenantService(tenantSubscribedEvent.getServiceName())) {
+ addTenantSubscriptionToLbContext(tenantSubscribedEvent.getServiceName(), tenantSubscribedEvent.getTenantId());
+ }
+ }
+ });
+ tenantEventReceiver.addEventListener(new TenantUnSubscribedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+ TenantUnSubscribedEvent tenantUnSubscribedEvent = (TenantUnSubscribedEvent) event;
+ if(log.isDebugEnabled()) {
+ log.debug(String.format("Tenant un-subscribed event received: [tenant-id] %d [service] %s",
+ tenantUnSubscribedEvent.getTenantId(), tenantUnSubscribedEvent.getServiceName()));
+ }
+ if(isMultiTenantService(tenantUnSubscribedEvent.getServiceName())) {
+ removeTenantSubscriptionFromLbContext(tenantUnSubscribedEvent.getServiceName(), tenantUnSubscribedEvent.getTenantId());
+ }
+ }
+ });
+ }
+
+ private boolean isMultiTenantService(String serviceName) {
+ try {
+ TopologyManager.acquireReadLock();
+ Service service = TopologyManager.getTopology().getService(serviceName);
+ if(service != null) {
+ return (service.getServiceType() == ServiceType.MultiTenant);
+ }
+ return false;
+ }
+ finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+
+ private void addTenantSubscriptionToLbContext(String serviceName, int tenantId) {
+ // Find cluster of tenant
+ Cluster cluster = findCluster(serviceName, tenantId);
+ if (cluster != null) {
+ for (String hostName : cluster.getHostNames()) {
+ // Add hostName, tenantId, cluster to multi-tenant map
+ Map<Integer, Cluster> clusterMap = LoadBalancerContext.getInstance().getMultiTenantClusterMap().getClusters(hostName);
+ if (clusterMap == null) {
+ clusterMap = new HashMap<Integer, Cluster>();
+ clusterMap.put(tenantId, cluster);
+ LoadBalancerContext.getInstance().getMultiTenantClusterMap().addClusters(hostName, clusterMap);
+ } else {
+ clusterMap.put(tenantId, cluster);
+ }
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Cluster added to multi-tenant cluster map: [host-name] %s [tenant-id] %d [cluster] %s",
+ hostName, tenantId, cluster.getClusterId()));
+ }
+ }
+ } else {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Could not find cluster of tenant: [service] %s [tenant-id] %d",
+ serviceName, tenantId));
+ }
+ }
+ }
+
+ private void removeTenantSubscriptionFromLbContext(String serviceName, int tenantId) {
+ // Find cluster of tenant
+ Cluster cluster = findCluster(serviceName, tenantId);
+ if (cluster != null) {
+ for (String hostName : cluster.getHostNames()) {
+ LoadBalancerContext.getInstance().getMultiTenantClusterMap().removeClusters(hostName);
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Cluster removed from multi-tenant clusters map: [host-name] %s [tenant-id] %d [cluster] %s",
+ hostName, tenantId, cluster.getClusterId()));
+ }
+ }
+ } else {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Could not find cluster of tenant: [service] %s [tenant-id] %d",
+ serviceName, tenantId));
+ }
+ }
+ }
+
+ private Cluster findCluster(String serviceName, int tenantId) {
+ try {
+ TopologyManager.acquireReadLock();
+ Service service = TopologyManager.getTopology().getService(serviceName);
+ if (service == null) {
+ throw new RuntimeException(String.format("Service not found: %s", serviceName));
+ }
+ for (Cluster cluster : service.getClusters()) {
+ if (cluster.tenantIdInRange(tenantId)) {
+ return cluster;
+ }
+ }
+ return null;
+ } finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+
+ @Override
+ public void run() {
+ Thread tenantReceiverThread = new Thread(tenantEventReceiver);
+ tenantReceiverThread.start();
+
+ // Keep the thread live until terminated
+ while (!terminated) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException ignore) {
+ }
+ }
+ if (log.isInfoEnabled()) {
+ log.info("Load balancer tenant receiver thread terminated");
+ }
+ }
+
+ /**
+ * Terminate load balancer tenant receiver thread.
+ */
+ public void terminate() {
+ tenantEventReceiver.terminate();
+ terminated = true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/bccad5be/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerTenantReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerTenantReceiver.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerTenantReceiver.java
deleted file mode 100644
index 316542f..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerTenantReceiver.java
+++ /dev/null
@@ -1,210 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.stratos.load.balancer;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.load.balancer.context.LoadBalancerContext;
-import org.apache.stratos.messaging.domain.tenant.Tenant;
-import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.Service;
-import org.apache.stratos.messaging.domain.topology.ServiceType;
-import org.apache.stratos.messaging.event.Event;
-import org.apache.stratos.messaging.event.tenant.CompleteTenantEvent;
-import org.apache.stratos.messaging.event.tenant.TenantSubscribedEvent;
-import org.apache.stratos.messaging.event.tenant.TenantUnSubscribedEvent;
-import org.apache.stratos.messaging.listener.tenant.CompleteTenantEventListener;
-import org.apache.stratos.messaging.listener.tenant.TenantSubscribedEventListener;
-import org.apache.stratos.messaging.listener.tenant.TenantUnSubscribedEventListener;
-import org.apache.stratos.messaging.message.processor.tenant.TenantMessageProcessorChain;
-import org.apache.stratos.messaging.message.receiver.tenant.TenantEventMessageDelegator;
-import org.apache.stratos.messaging.message.receiver.tenant.TenantReceiver;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * Load balancer tenant receiver updates load balancer context according to
- * incoming tenant events.
- */
-public class LoadBalancerTenantReceiver implements Runnable {
-
- private static final Log log = LogFactory.getLog(LoadBalancerTenantReceiver.class);
-
- private final TenantReceiver tenantReceiver;
- private boolean terminated;
-
- public LoadBalancerTenantReceiver() {
- tenantReceiver = new TenantReceiver(createMessageDelegator());
- }
-
- private TenantEventMessageDelegator createMessageDelegator() {
- TenantMessageProcessorChain processorChain = createEventProcessorChain();
- return new TenantEventMessageDelegator(processorChain);
- }
-
- private TenantMessageProcessorChain createEventProcessorChain() {
- TenantMessageProcessorChain messageProcessorChain = new TenantMessageProcessorChain();
- messageProcessorChain.addEventListener(new CompleteTenantEventListener() {
- @Override
- protected void onEvent(Event event) {
- CompleteTenantEvent completeTenantEvent = (CompleteTenantEvent) event;
- for (Tenant tenant : completeTenantEvent.getTenants()) {
- for (String serviceName : tenant.getServiceSubscriptions()) {
- if(isMultiTenantService(serviceName)) {
- addTenantSubscriptionToLbContext(serviceName, tenant.getTenantId());
- }
- }
- }
- }
- });
- messageProcessorChain.addEventListener(new TenantSubscribedEventListener() {
- @Override
- protected void onEvent(Event event) {
- TenantSubscribedEvent tenantSubscribedEvent = (TenantSubscribedEvent) event;
- if(log.isDebugEnabled()) {
- log.debug(String.format("Tenant subscribed event received: [tenant-id] %d [service] %s",
- tenantSubscribedEvent.getTenantId(), tenantSubscribedEvent.getServiceName()));
- }
- if(isMultiTenantService(tenantSubscribedEvent.getServiceName())) {
- addTenantSubscriptionToLbContext(tenantSubscribedEvent.getServiceName(), tenantSubscribedEvent.getTenantId());
- }
- }
- });
- messageProcessorChain.addEventListener(new TenantUnSubscribedEventListener() {
- @Override
- protected void onEvent(Event event) {
- TenantUnSubscribedEvent tenantUnSubscribedEvent = (TenantUnSubscribedEvent) event;
- if(log.isDebugEnabled()) {
- log.debug(String.format("Tenant un-subscribed event received: [tenant-id] %d [service] %s",
- tenantUnSubscribedEvent.getTenantId(), tenantUnSubscribedEvent.getServiceName()));
- }
- if(isMultiTenantService(tenantUnSubscribedEvent.getServiceName())) {
- removeTenantSubscriptionFromLbContext(tenantUnSubscribedEvent.getServiceName(), tenantUnSubscribedEvent.getTenantId());
- }
- }
- });
- return messageProcessorChain;
- }
-
- private boolean isMultiTenantService(String serviceName) {
- try {
- TopologyManager.acquireReadLock();
- Service service = TopologyManager.getTopology().getService(serviceName);
- if(service != null) {
- return (service.getServiceType() == ServiceType.MultiTenant);
- }
- return false;
- }
- finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- private void addTenantSubscriptionToLbContext(String serviceName, int tenantId) {
- // Find cluster of tenant
- Cluster cluster = findCluster(serviceName, tenantId);
- if (cluster != null) {
- for (String hostName : cluster.getHostNames()) {
- // Add hostName, tenantId, cluster to multi-tenant map
- Map<Integer, Cluster> clusterMap = LoadBalancerContext.getInstance().getMultiTenantClusterMap().getClusters(hostName);
- if (clusterMap == null) {
- clusterMap = new HashMap<Integer, Cluster>();
- clusterMap.put(tenantId, cluster);
- LoadBalancerContext.getInstance().getMultiTenantClusterMap().addClusters(hostName, clusterMap);
- } else {
- clusterMap.put(tenantId, cluster);
- }
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster added to multi-tenant cluster map: [host-name] %s [tenant-id] %d [cluster] %s",
- hostName, tenantId, cluster.getClusterId()));
- }
- }
- } else {
- if (log.isErrorEnabled()) {
- log.error(String.format("Could not find cluster of tenant: [service] %s [tenant-id] %d",
- serviceName, tenantId));
- }
- }
- }
-
- private void removeTenantSubscriptionFromLbContext(String serviceName, int tenantId) {
- // Find cluster of tenant
- Cluster cluster = findCluster(serviceName, tenantId);
- if (cluster != null) {
- for (String hostName : cluster.getHostNames()) {
- LoadBalancerContext.getInstance().getMultiTenantClusterMap().removeClusters(hostName);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster removed from multi-tenant clusters map: [host-name] %s [tenant-id] %d [cluster] %s",
- hostName, tenantId, cluster.getClusterId()));
- }
- }
- } else {
- if (log.isErrorEnabled()) {
- log.error(String.format("Could not find cluster of tenant: [service] %s [tenant-id] %d",
- serviceName, tenantId));
- }
- }
- }
-
- private Cluster findCluster(String serviceName, int tenantId) {
- try {
- TopologyManager.acquireReadLock();
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- throw new RuntimeException(String.format("Service not found: %s", serviceName));
- }
- for (Cluster cluster : service.getClusters()) {
- if (cluster.tenantIdInRange(tenantId)) {
- return cluster;
- }
- }
- return null;
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- @Override
- public void run() {
- Thread tenantReceiverThread = new Thread(tenantReceiver);
- tenantReceiverThread.start();
-
- // Keep the thread live until terminated
- while (!terminated) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ignore) {
- }
- }
- if (log.isInfoEnabled()) {
- log.info("Load balancer tenant receiver thread terminated");
- }
- }
-
- /**
- * Terminate load balancer tenant receiver thread.
- */
- public void terminate() {
- tenantReceiver.terminate();
- terminated = true;
- }
-}