You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stratos.apache.org by re...@apache.org on 2014/09/16 10:47:44 UTC
[2/2] git commit: adding monitor support for grouping feature
adding monitor support for grouping feature
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/1422611d
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/1422611d
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/1422611d
Branch: refs/heads/4.0.0-grouping
Commit: 1422611dec3a618adf711ddaea66d5e551130316
Parents: 1bfee8b
Author: Reka <re...@Rekas-MacBook-Pro.local>
Authored: Tue Sep 16 14:15:52 2014 +0530
Committer: Reka <re...@Rekas-MacBook-Pro.local>
Committed: Tue Sep 16 14:15:52 2014 +0530
----------------------------------------------------------------------
.../stratos/autoscaler/AutoscalerContext.java | 42 +-
.../autoscaler/grouping/DependencyBuilder.java | 26 ++
.../AutoscalerHealthStatEventReceiver.java | 27 +-
.../AutoscalerTopologyEventReceiver.java | 468 ++++++-------------
.../monitor/AbstractClusterMonitor.java | 223 +++++++++
.../autoscaler/monitor/AbstractMonitor.java | 227 ---------
.../autoscaler/monitor/ClusterMonitor.java | 244 ----------
.../monitor/CompositeApplicationMonitor.java | 254 ----------
.../autoscaler/monitor/LbClusterMonitor.java | 125 -----
.../stratos/autoscaler/monitor/Monitor.java | 281 +++++++++++
.../monitor/application/ApplicationMonitor.java | 142 ++++++
.../monitor/cluster/ClusterMonitor.java | 276 +++++++++++
.../monitor/cluster/LbClusterMonitor.java | 128 +++++
.../autoscaler/monitor/group/GroupMonitor.java | 48 ++
.../status/checker/ClusterStatusChecker.java | 39 ++
.../status/checker/GroupStatusChecker.java | 37 ++
.../status/checker/StatusChecker.java | 48 ++
.../stratos/autoscaler/util/AutoscalerUtil.java | 26 +-
18 files changed, 1460 insertions(+), 1201 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/1422611d/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/AutoscalerContext.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/AutoscalerContext.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/AutoscalerContext.java
index ed54eb8..4ea6b28 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/AutoscalerContext.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/AutoscalerContext.java
@@ -22,9 +22,10 @@ package org.apache.stratos.autoscaler;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.monitor.ClusterMonitor;
-import org.apache.stratos.autoscaler.monitor.CompositeApplicationMonitor;
-import org.apache.stratos.autoscaler.monitor.LbClusterMonitor;
+import org.apache.stratos.autoscaler.monitor.application.ApplicationMonitor;
+import org.apache.stratos.autoscaler.monitor.cluster.ClusterMonitor;
+import org.apache.stratos.autoscaler.monitor.group.GroupMonitor;
+import org.apache.stratos.autoscaler.monitor.cluster.LbClusterMonitor;
import java.util.HashMap;
import java.util.Map;
@@ -49,20 +50,30 @@ public class AutoscalerContext {
// Map<LBClusterId, LBClusterMonitor>
private Map<String, LbClusterMonitor> lbMonitors;
- private Map<String, CompositeApplicationMonitor> appMonitors;
+ private Map<String, GroupMonitor> groupMonitors;
- public Map<String, CompositeApplicationMonitor> getAppMonitors() {
+ private Map<String, ApplicationMonitor> appMonitors;
+
+ public Map<String, ApplicationMonitor> getAppMonitors() {
return appMonitors;
}
- public CompositeApplicationMonitor getAppMonitor(String applicationId) {
+ public ApplicationMonitor getAppMonitor(String applicationId) {
return appMonitors.get(applicationId);
}
- public void setAppMonitors(Map<String, CompositeApplicationMonitor> appMonitors) {
+ public void setAppMonitors(Map<String, ApplicationMonitor> appMonitors) {
this.appMonitors = appMonitors;
}
+ public Map<String, GroupMonitor> getGroupMonitors() {
+ return groupMonitors;
+ }
+
+ public void setGroupMonitors(Map<String, GroupMonitor> groupMonitors) {
+ this.groupMonitors = groupMonitors;
+ }
+
private static class Holder {
private static final AutoscalerContext INSTANCE = new AutoscalerContext();
@@ -83,6 +94,14 @@ public class AutoscalerContext {
public boolean monitorExist(String clusterId) {
return monitors.containsKey(clusterId);
}
+
+ public void addAppMonitor(ApplicationMonitor appMonitor) {
+ appMonitors.put(appMonitor.getId(), appMonitor);
+ }
+
+ public boolean appMonitorExist(String appId) {
+ return appMonitors.containsKey(appId);
+ }
public boolean lbMonitorExist(String clusterId) {
return lbMonitors.containsKey(clusterId);
@@ -126,11 +145,7 @@ public class AutoscalerContext {
lbMonitors.put(monitor.getClusterId(), monitor);
}
- public void addAppMonitor(CompositeApplicationMonitor monitor) {
- appMonitors.put(monitor.getAppId(), monitor);
- }
-
- public CompositeApplicationMonitor removeAppMonitor(String appId) {
+ public ApplicationMonitor removeAppMonitor(String appId) {
if(!appMonitorExist(appId)) {
log.fatal("LB monitor not found for App id: "+ appId);
return null;
@@ -139,7 +154,4 @@ public class AutoscalerContext {
return appMonitors.remove(appId);
}
- public boolean appMonitorExist(String appId) {
- return appMonitors.containsKey(appId);
- }
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/1422611d/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/grouping/DependencyBuilder.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/grouping/DependencyBuilder.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/grouping/DependencyBuilder.java
new file mode 100644
index 0000000..0566e49
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/grouping/DependencyBuilder.java
@@ -0,0 +1,26 @@
+/*
+ * 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.grouping;
+
+/**
+ * This is to build the startup/termination dependencies
+ * across all the groups and clusters
+ */
+public class DependencyBuilder {
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/1422611d/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/AutoscalerHealthStatEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/AutoscalerHealthStatEventReceiver.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/AutoscalerHealthStatEventReceiver.java
index 52952c3..9ffe66d 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/AutoscalerHealthStatEventReceiver.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/AutoscalerHealthStatEventReceiver.java
@@ -26,7 +26,7 @@ import org.apache.stratos.autoscaler.NetworkPartitionContext;
import org.apache.stratos.autoscaler.PartitionContext;
import org.apache.stratos.autoscaler.client.cloud.controller.CloudControllerClient;
import org.apache.stratos.autoscaler.exception.TerminationException;
-import org.apache.stratos.autoscaler.monitor.AbstractMonitor;
+import org.apache.stratos.autoscaler.monitor.AbstractClusterMonitor;
import org.apache.stratos.autoscaler.policy.model.LoadAverage;
import org.apache.stratos.autoscaler.policy.model.MemoryConsumption;
import org.apache.stratos.messaging.domain.topology.Cluster;
@@ -94,7 +94,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -136,7 +136,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -178,7 +178,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -218,7 +218,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -259,7 +259,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -299,7 +299,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -340,7 +340,6 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
, floatValue));
}
}
-
}
});
@@ -461,7 +460,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -502,7 +501,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -542,7 +541,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
clusterId, networkPartitionId, floatValue));
}
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -584,7 +583,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
String clusterId = member.getClusterId();
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
@@ -628,7 +627,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
}
return null;
}
- AbstractMonitor monitor = AutoscalerContext.getInstance().getMonitor(member.getClusterId());
+ AbstractClusterMonitor monitor = AutoscalerContext.getInstance().getMonitor(member.getClusterId());
if(null == monitor){
monitor = AutoscalerContext.getInstance().getLBMonitor(member.getClusterId());
@@ -692,7 +691,7 @@ public class AutoscalerHealthStatEventReceiver implements Runnable {
private void handleMemberFaultEvent(String clusterId, String memberId) {
try {
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if(asCtx.monitorExist(clusterId)){
monitor = asCtx.getMonitor(clusterId);
http://git-wip-us.apache.org/repos/asf/stratos/blob/1422611d/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
index 2ac006b..9ada44e 100644
--- 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
@@ -24,26 +24,20 @@ import org.apache.commons.logging.LogFactory;
import org.apache.stratos.autoscaler.*;
import org.apache.stratos.autoscaler.client.cloud.controller.CloudControllerClient;
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.exception.TerminationException;
-import org.apache.stratos.autoscaler.monitor.AbstractMonitor;
-import org.apache.stratos.autoscaler.monitor.CompositeApplicationMonitor;
+import org.apache.stratos.autoscaler.monitor.AbstractClusterMonitor;
+import org.apache.stratos.autoscaler.monitor.application.ApplicationMonitor;
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.Application;
import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.CompositeApplication;
-import org.apache.stratos.messaging.domain.topology.ConfigCompositeApplication;
import org.apache.stratos.messaging.domain.topology.Service;
-import org.apache.stratos.messaging.domain.topology.util.CompositeApplicationBuilder;
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;
@@ -92,11 +86,10 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
topologyEventReceiver.addEventListener(new CompleteTopologyEventListener() {
@Override
protected void onEvent(Event event) {
-
try {
TopologyManager.acquireReadLock();
- for (CompositeApplication compositeApplication : TopologyManager.getTopology().getCompositeApplication()) {
-
+ for (Application application : TopologyManager.getTopology().getApplications()) {
+ startApplicationMonitor(application);
}
} catch (Exception e) {
log.error("Error processing event", e);
@@ -109,27 +102,22 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
});
- topologyEventReceiver.addEventListener(new CompositeApplicationCreatedEventListener() {
+ topologyEventReceiver.addEventListener(new ApplicationCreatedEventListener() {
@Override
protected void onEvent(Event event) {
- log.info("[ClusterCreatedEventListener] Received: " + event.getClass());
-
- CompositeApplicationCreatedEvent compositeApplicationCreatedEvent = (CompositeApplicationCreatedEvent) event;
+ log.info("[ApplicationCreatedEventListener] Received: " + event.getClass());
- ConfigCompositeApplication configCompositeApplication =
- compositeApplicationCreatedEvent.getCompositeApplication();
+ ApplicationCreatedEvent applicationCreatedEvent = (ApplicationCreatedEvent) event;
//acquire read lock
TopologyManager.acquireReadLock();
try {
- CompositeApplicationBuilder builder = new CompositeApplicationBuilder();
- CompositeApplication compositeApplication =
- builder.buildCompositeApplication(TopologyManager.getTopology(),
- configCompositeApplication.getAlias());
- //start the app monitor
+ //TODO build dependency and organize the application
+ //start the application monitor
+ startApplicationMonitor(applicationCreatedEvent.getApplication());
} finally {
//release read lock
@@ -139,48 +127,56 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
}
});
-// 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 ApplicationRemovedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ log.info("[ApplicationCreatedEventListener] Received: " + event.getClass());
+
+ ApplicationRemovedEvent applicationCreatedEvent = (ApplicationRemovedEvent) event;
+
+ //acquire read lock
+ TopologyManager.acquireReadLock();
+
+ try {
+ //TODO build dependency and organize the application
+
+ //remove monitors by checking the termination
+
+ } finally {
+ //release read lock
+ TopologyManager.releaseReadLock();
+ }
+
+ }
+ });
topologyEventReceiver.addEventListener(new MemberReadyToShutdownEventListener() {
@Override
protected void onEvent(Event event) {
try {
- MemberReadyToShutdownEvent memberReadyToShutdownEvent = (MemberReadyToShutdownEvent)event;
+ MemberReadyToShutdownEvent memberReadyToShutdownEvent =
+ (MemberReadyToShutdownEvent) event;
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
String clusterId = memberReadyToShutdownEvent.getClusterId();
String memberId = memberReadyToShutdownEvent.getMemberId();
- if(asCtx.monitorExist(clusterId)){
+ if (asCtx.monitorExist(clusterId)) {
monitor = asCtx.getMonitor(clusterId);
- }else if(asCtx.lbMonitorExist(clusterId)){
+ } else if (asCtx.lbMonitorExist(clusterId)) {
monitor = asCtx.getLBMonitor(clusterId);
- }else{
- if(log.isDebugEnabled()){
- log.debug(String.format("A cluster monitor is not found in autoscaler context [cluster] %s", clusterId));
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("A cluster monitor is not found " +
+ "in autoscaler context [cluster] %s", clusterId));
}
return;
}
NetworkPartitionContext nwPartitionCtxt;
- nwPartitionCtxt = monitor.getNetworkPartitionCtxt(memberReadyToShutdownEvent.getNetworkPartitionId());
+ nwPartitionCtxt = monitor.getNetworkPartitionCtxt(
+ memberReadyToShutdownEvent.getNetworkPartitionId());
// start a new member in the same Partition
String partitionId = monitor.getPartitionOfMember(memberId);
@@ -195,7 +191,8 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
partitionCtxt.removeActiveMemberById(memberId);
if (log.isInfoEnabled()) {
- log.info(String.format("Member is terminated and removed from the active members list: [member] %s [partition] %s [cluster] %s ",
+ log.info(String.format("Member is terminated and removed from the active " +
+ "members list: [member] %s [partition] %s [cluster] %s ",
memberId, partitionId, clusterId));
}
} catch (TerminationException e) {
@@ -215,10 +212,13 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
TopologyManager.acquireReadLock();
Service service = TopologyManager.getTopology().getService(e.getServiceName());
Cluster cluster = service.getCluster(e.getClusterId());
- if(AutoscalerContext.getInstance().monitorExist((cluster.getClusterId()))) {
- AutoscalerContext.getInstance().getMonitor(e.getClusterId()).setStatus(e.getStatus());
- } else if (AutoscalerContext.getInstance().lbMonitorExist((cluster.getClusterId()))) {
- AutoscalerContext.getInstance().getLBMonitor(e.getClusterId()).setStatus(e.getStatus());
+ if (AutoscalerContext.getInstance().monitorExist((cluster.getClusterId()))) {
+ AutoscalerContext.getInstance().getMonitor(e.getClusterId()).
+ setStatus(e.getStatus());
+ } else if (AutoscalerContext.getInstance().
+ lbMonitorExist((cluster.getClusterId()))) {
+ AutoscalerContext.getInstance().getLBMonitor(e.getClusterId()).
+ setStatus(e.getStatus());
} else {
log.error("cluster monitor not exists for the cluster: " + cluster.toString());
}
@@ -242,10 +242,11 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
String clusterId = e.getClusterId();
String deploymentPolicy = e.getDeploymentPolicy();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if (e.isLbCluster()) {
- DeploymentPolicy depPolicy = PolicyManager.getInstance().getDeploymentPolicy(deploymentPolicy);
+ DeploymentPolicy depPolicy = PolicyManager.getInstance().
+ getDeploymentPolicy(deploymentPolicy);
if (depPolicy != null) {
List<NetworkPartitionLbHolder> lbHolders = PartitionManager.getInstance()
.getNetworkPartitionLbHolders(depPolicy);
@@ -308,7 +309,7 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
String networkPartitionId = e.getNetworkPartitionId();
String clusterId = e.getClusterId();
String partitionId = e.getPartitionId();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if (AutoscalerContext.getInstance().monitorExist(clusterId)) {
monitor = AutoscalerContext.getInstance().getMonitor(clusterId);
@@ -317,28 +318,35 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
monitor = AutoscalerContext.getInstance().getLBMonitor(clusterId);
}
- NetworkPartitionContext networkPartitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId);
+ NetworkPartitionContext networkPartitionContext = monitor.
+ getNetworkPartitionCtxt(networkPartitionId);
- PartitionContext partitionContext = networkPartitionContext.getPartitionCtxt(partitionId);
+ 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));
+ 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));
+ 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));
+ 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));
+ 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));
+ log.info(String.format("Member stat context has been removed " +
+ " successfully: [member] %s", memberId));
}
// partitionContext.decrementCurrentActiveMemberCount(1);
@@ -361,37 +369,16 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
MemberActivatedEvent e = (MemberActivatedEvent) event;
String memberId = e.getMemberId();
- String partitionId = e.getPartitionId();
- String networkPartitionId = e.getNetworkPartitionId();
- String applicationId = e.getApplicationId();
PartitionContext partitionContext = null;
- String clusterId = e.getClusterId();
- CompositeApplicationMonitor appMonitor;
- AbstractMonitor clusterMonitor;
-
- if (AutoscalerContext.getInstance().appMonitorExist(applicationId)) {
- appMonitor = AutoscalerContext.getInstance().getAppMonitor(applicationId);
- if((appMonitor).clusterMonitorExists(clusterId)) {
- clusterMonitor = appMonitor.getClusterMonitor(clusterId);
- partitionContext = clusterMonitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
- }
- else if(appMonitor.lbMonitorExists(clusterId)) {
- clusterMonitor = appMonitor.getLBclusterMonitor(clusterId);
- partitionContext = clusterMonitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
- } else {
- log.error(String.format("Couldn't find the Cluster [monitor] %s for the [member] %s", clusterId, memberId));
- return;
- }
- partitionContext.addMemberStatsContext(new MemberStatsContext(memberId));
- //starting the pending clusters which are waiting for this member activation in a cluster
- appMonitor.registerClusterMonitor();
- } else {
- log.error(String.format("Couldn't find the Application [monitor] %s for the [member] %s", applicationId, memberId));
- return;
- }
+
+ partitionContext.addMemberStatsContext(new MemberStatsContext(memberId));
+ // TODO starting the pending clusters which are waiting for this member activation in a cluster
+
+
if (log.isInfoEnabled()) {
- log.info(String.format("Member stat context has been added successfully: [member] %s", memberId));
+ log.info(String.format("Member stat context has been added " +
+ "successfully: [member] %s", memberId));
}
// partitionContext.incrementCurrentActiveMemberCount(1);
partitionContext.movePendingMemberToActiveMembers(memberId);
@@ -408,25 +395,28 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
@Override
protected void onEvent(Event event) {
try {
- MemberReadyToShutdownEvent memberReadyToShutdownEvent = (MemberReadyToShutdownEvent)event;
+ MemberReadyToShutdownEvent memberReadyToShutdownEvent =
+ (MemberReadyToShutdownEvent) event;
AutoscalerContext asCtx = AutoscalerContext.getInstance();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
String clusterId = memberReadyToShutdownEvent.getClusterId();
String memberId = memberReadyToShutdownEvent.getMemberId();
- if(asCtx.monitorExist(clusterId)){
+ if (asCtx.monitorExist(clusterId)) {
monitor = asCtx.getMonitor(clusterId);
- }else if(asCtx.lbMonitorExist(clusterId)){
+ } else if (asCtx.lbMonitorExist(clusterId)) {
monitor = asCtx.getLBMonitor(clusterId);
- }else{
- if(log.isDebugEnabled()){
- log.debug(String.format("A cluster monitor is not found in autoscaler context [cluster] %s", clusterId));
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("A cluster monitor is not found in autoscaler " +
+ "context [cluster] %s", clusterId));
}
return;
}
NetworkPartitionContext nwPartitionCtxt;
- nwPartitionCtxt = monitor.getNetworkPartitionCtxt(memberReadyToShutdownEvent.getNetworkPartitionId());
+ nwPartitionCtxt = monitor.getNetworkPartitionCtxt(memberReadyToShutdownEvent.
+ getNetworkPartitionId());
// start a new member in the same Partition
String partitionId = monitor.getPartitionOfMember(memberId);
@@ -441,7 +431,8 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
partitionCtxt.removeActiveMemberById(memberId);
if (log.isInfoEnabled()) {
- log.info(String.format("Member is terminated and removed from the active members list: [member] %s [partition] %s [cluster] %s ",
+ log.info(String.format("Member is terminated and removed from the active " +
+ "members list: [member] %s [partition] %s [cluster] %s ",
memberId, partitionId, clusterId));
}
} catch (TerminationException e) {
@@ -466,18 +457,21 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
PartitionContext partitionContext;
String clusterId = e.getClusterId();
- AbstractMonitor monitor;
+ AbstractClusterMonitor monitor;
if (AutoscalerContext.getInstance().monitorExist(clusterId)) {
monitor = AutoscalerContext.getInstance().getMonitor(clusterId);
- partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
+ partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).
+ getPartitionCtxt(partitionId);
} else {
monitor = AutoscalerContext.getInstance().getLBMonitor(clusterId);
- partitionContext = monitor.getNetworkPartitionCtxt(networkPartitionId).getPartitionCtxt(partitionId);
+ 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));
+ log.debug(String.format("Member has been moved as pending termination: " +
+ "[member] %s", memberId));
}
partitionContext.moveActiveMemberToTerminationPendingMembers(memberId);
@@ -488,93 +482,48 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
}
}
});
+ }
- 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();
-// }
- }
- });
-
-
-
+ /**
+ * Terminate load balancer topology receiver thread.
+ */
+ public void terminate() {
+ topologyEventReceiver.terminate();
+ terminated = true;
+ }
+ protected synchronized void startApplicationMonitor(Application application) {
+ Thread th = null;
+ if (!AutoscalerContext.getInstance()
+ .appMonitorExist(application.getId())) {
+ th = new Thread(
+ new ApplicationMonitorAdder(application));
+ }
+ if (th != null) {
+ th.start();
+ try {
+ th.join();
+ } catch (InterruptedException ignore) {
+ }
+ if (log.isDebugEnabled()) {
+ log.debug(String
+ .format("Application monitor thread has been started successfully: " +
+ "[application] %s ", application.getId()));
+ }
+ }
}
-// 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 AppMonitorAdder implements Runnable {
- private CompositeApplication compositeApplication;
-
- public AppMonitorAdder(CompositeApplication compositeApplication) {
- this.compositeApplication = compositeApplication;
+ private class ApplicationMonitorAdder implements Runnable {
+ private Application application;
+
+ public ApplicationMonitorAdder(Application application) {
+ this.application = application;
}
public void run() {
- CompositeApplicationMonitor monitor = null;
+ ApplicationMonitor applicationMonitor = null;
int retries = 5;
boolean success = false;
do {
@@ -582,149 +531,36 @@ public class AutoscalerTopologyEventReceiver implements Runnable {
Thread.sleep(5000);
} catch (InterruptedException e1) {
}
- monitor = new CompositeApplicationMonitor(compositeApplication);
- success = true;
- } while (!success && retries <= 0);
+ try {
+ applicationMonitor = AutoscalerUtil.getApplicationMonitor(application);
+ success = true;
+ //TODO exception handling
+ } catch (Exception e) {
+ String msg = "Application monitor creation failed for Application: " +
+ application.getId();
+ log.debug(msg, e);
+ retries--;
+
+ }
+ } while (!success && retries != 0);
- if (monitor == null) {
- String msg = "App monitor creation failed, even after retrying for 5 times, "
- + "for Composite Application: " + compositeApplication.getAlias();
+ if (applicationMonitor == null) {
+ String msg = "Application monitor creation failed, even after retrying for 5 times, "
+ + "for Application: " + applicationMonitor.getId();
log.error(msg);
throw new RuntimeException(msg);
}
- Thread th = new Thread(monitor);
+ Thread th = new Thread(applicationMonitor);
th.start();
- AutoscalerContext.getInstance().addAppMonitor(monitor);
+
if (log.isInfoEnabled()) {
- log.info(String.format("App monitor has been added successfully: [Composite Application] %s",
- compositeApplication.getAlias()));
+ log.info(String.format("Application monitor has been added successfully: " +
+ "[application] %s", applicationMonitor.getId()));
}
}
}
-//
-//
-// 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()));
-// }
-// }
-// }
-
- @SuppressWarnings("unused")
- 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()));
-// }
-// }
-// }
-
- protected synchronized void startAppMonitor(CompositeApplication compositeApplication) {
- Thread th = null;
- if (AutoscalerContext.getInstance()
- .appMonitorExist(
- compositeApplication.getAlias())) {
- th = new Thread(new AppMonitorAdder(
- compositeApplication));
- }
- if (th != null) {
- th.start();
- try {
- th.join();
- } catch (InterruptedException ignore) {
- }
-
- if (log.isDebugEnabled()) {
- log.debug(String
- .format("Composite Application monitor thread has been started successfully: [Composite Application] %s ",
- compositeApplication.getAlias()));
- }
- }
- }
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/1422611d/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/AbstractClusterMonitor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/AbstractClusterMonitor.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/AbstractClusterMonitor.java
new file mode 100644
index 0000000..4bf8b7f
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/AbstractClusterMonitor.java
@@ -0,0 +1,223 @@
+/*
+ * 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.monitor;
+
+import org.apache.stratos.autoscaler.policy.model.AutoscalePolicy;
+
+import java.util.Map;
+
+import org.apache.commons.configuration.XMLConfiguration;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.autoscaler.NetworkPartitionContext;
+import org.apache.stratos.autoscaler.deployment.policy.DeploymentPolicy;
+import org.apache.stratos.autoscaler.rule.AutoscalerRuleEvaluator;
+import org.apache.stratos.autoscaler.util.AutoScalerConstants;
+import org.apache.stratos.autoscaler.util.ConfUtil;
+import org.apache.stratos.messaging.domain.topology.Cluster;
+import org.apache.stratos.messaging.domain.topology.Member;
+import org.apache.stratos.messaging.domain.topology.Service;
+import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
+import org.drools.runtime.StatefulKnowledgeSession;
+import org.drools.runtime.rule.FactHandle;
+
+/**
+ * Is responsible for monitoring a service cluster. This runs periodically
+ * and perform minimum instance check and scaling check using the underlying
+ * rules engine.
+ */
+abstract public class AbstractClusterMonitor implements Runnable {
+
+ private static final Log log = LogFactory.getLog(AbstractClusterMonitor.class);
+ // Map<NetworkpartitionId, Network Partition Context>
+ protected Map<String, NetworkPartitionContext> networkPartitionCtxts;
+ protected DeploymentPolicy deploymentPolicy;
+ protected AutoscalePolicy autoscalePolicy;
+
+
+ protected FactHandle minCheckFactHandle;
+ protected FactHandle scaleCheckFactHandle;
+ protected FactHandle terminateDependencyFactHandle;
+
+ protected StatefulKnowledgeSession minCheckKnowledgeSession;
+ protected StatefulKnowledgeSession scaleCheckKnowledgeSession;
+ protected StatefulKnowledgeSession terminateDependencyKnowledgeSession;
+ protected boolean isDestroyed;
+
+ protected String clusterId;
+ protected String serviceId;
+
+ protected AutoscalerRuleEvaluator autoscalerRuleEvaluator;
+
+ // time intereval between two runs of the Monitor. Default is 90000ms.
+ protected int monitorInterval;
+
+ public AbstractClusterMonitor() {
+ readConfigurations();
+ }
+
+ private void readConfigurations () {
+
+ XMLConfiguration conf = ConfUtil.getInstance(null).getConfiguration();
+ monitorInterval = conf.getInt(AutoScalerConstants.AUTOSCALER_MONITOR_INTERVAL, 90000);
+ if (log.isDebugEnabled()) {
+ log.debug("Cluster Monitor task interval: " + getMonitorInterval());
+ }
+ }
+
+ @Override
+ public void run() {
+ // TODO Auto-generated method stub
+ }
+
+
+ public NetworkPartitionContext getNetworkPartitionCtxt(Member member) {
+ log.info("***** getNetworkPartitionCtxt " + member.getNetworkPartitionId());
+ String networkPartitionId = member.getNetworkPartitionId();
+ if(networkPartitionCtxts.containsKey(networkPartitionId)) {
+ log.info("returnnig network partition context " + networkPartitionCtxts.get(networkPartitionId));
+ return networkPartitionCtxts.get(networkPartitionId);
+ }
+ log.info("returning null getNetworkPartitionCtxt");
+ return null;
+ }
+
+ public String getPartitionOfMember(String memberId){
+ for(Service service: TopologyManager.getTopology().getServices()){
+ for(Cluster cluster: service.getClusters()){
+ if(cluster.memberExists(memberId)){
+ return cluster.getMember(memberId).getPartitionId();
+ }
+ }
+ }
+ return null;
+ }
+
+ public void destroy() {
+ minCheckKnowledgeSession.dispose();
+ scaleCheckKnowledgeSession.dispose();
+ terminateDependencyKnowledgeSession.dispose();
+ setDestroyed(true);
+ if (log.isDebugEnabled()) {
+ log.debug("Cluster Monitor Drools session has been disposed. " + this.toString());
+ }
+ }
+
+ public boolean isDestroyed() {
+ return isDestroyed;
+ }
+
+ public void setDestroyed(boolean isDestroyed) {
+ this.isDestroyed = isDestroyed;
+ }
+
+ public String getServiceId() {
+ return serviceId;
+ }
+
+ public void setServiceId(String serviceId) {
+ this.serviceId = serviceId;
+ }
+
+ public DeploymentPolicy getDeploymentPolicy() {
+ return deploymentPolicy;
+ }
+
+ public void setDeploymentPolicy(DeploymentPolicy deploymentPolicy) {
+ this.deploymentPolicy = deploymentPolicy;
+ }
+
+ public AutoscalePolicy getAutoscalePolicy() {
+ return autoscalePolicy;
+ }
+
+ public void setAutoscalePolicy(AutoscalePolicy autoscalePolicy) {
+ this.autoscalePolicy = autoscalePolicy;
+ }
+
+ public String getClusterId() {
+ return clusterId;
+ }
+
+ public void setClusterId(String clusterId) {
+ this.clusterId = clusterId;
+ }
+
+ public Map<String, NetworkPartitionContext> getNetworkPartitionCtxts() {
+ return networkPartitionCtxts;
+ }
+
+ public NetworkPartitionContext getNetworkPartitionCtxt(String networkPartitionId) {
+ return networkPartitionCtxts.get(networkPartitionId);
+ }
+
+ public void setPartitionCtxt(Map<String, NetworkPartitionContext> partitionCtxt) {
+ this.networkPartitionCtxts = partitionCtxt;
+ }
+
+ public boolean partitionCtxtAvailable(String partitionId) {
+ return networkPartitionCtxts.containsKey(partitionId);
+ }
+
+ public void addNetworkPartitionCtxt(NetworkPartitionContext ctxt) {
+ this.networkPartitionCtxts.put(ctxt.getId(), ctxt);
+ }
+
+ public NetworkPartitionContext getPartitionCtxt(String id) {
+ return this.networkPartitionCtxts.get(id);
+ }
+
+ public StatefulKnowledgeSession getMinCheckKnowledgeSession() {
+ return minCheckKnowledgeSession;
+ }
+
+ public void setMinCheckKnowledgeSession(StatefulKnowledgeSession minCheckKnowledgeSession) {
+ this.minCheckKnowledgeSession = minCheckKnowledgeSession;
+ }
+
+ public FactHandle getMinCheckFactHandle() {
+ return minCheckFactHandle;
+ }
+
+ public void setMinCheckFactHandle(FactHandle minCheckFactHandle) {
+ this.minCheckFactHandle = minCheckFactHandle;
+ }
+
+ public StatefulKnowledgeSession getTerminateDependencyKnowledgeSession() {
+ return terminateDependencyKnowledgeSession;
+ }
+
+ public void setTerminateDependencyKnowledgeSession(
+ StatefulKnowledgeSession terminateDependencyKnowledgeSession) {
+ this.terminateDependencyKnowledgeSession = terminateDependencyKnowledgeSession;
+ }
+
+ public FactHandle getTerminateDependencyFactHandle() {
+ return terminateDependencyFactHandle;
+ }
+
+ public void setTerminateDependencyFactHandle(
+ FactHandle terminateDependencyFactHandle) {
+ this.terminateDependencyFactHandle = terminateDependencyFactHandle;
+ }
+
+ public int getMonitorInterval() {
+ return monitorInterval;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/stratos/blob/1422611d/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/AbstractMonitor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/AbstractMonitor.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/AbstractMonitor.java
deleted file mode 100644
index 3c95c91..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/AbstractMonitor.java
+++ /dev/null
@@ -1,227 +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.monitor;
-
-import org.apache.stratos.autoscaler.policy.model.AutoscalePolicy;
-
-import java.util.Map;
-
-import org.apache.commons.configuration.XMLConfiguration;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.NetworkPartitionContext;
-import org.apache.stratos.autoscaler.deployment.policy.DeploymentPolicy;
-import org.apache.stratos.autoscaler.policy.model.AutoscalePolicy;
-import org.apache.stratos.autoscaler.rule.AutoscalerRuleEvaluator;
-import org.apache.stratos.autoscaler.util.AutoScalerConstants;
-import org.apache.stratos.autoscaler.util.ConfUtil;
-import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.Member;
-import org.apache.stratos.messaging.domain.topology.Service;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
-import org.drools.runtime.StatefulKnowledgeSession;
-import org.drools.runtime.rule.FactHandle;
-
-import java.util.Map;
-
-/**
- * Is responsible for monitoring a service cluster. This runs periodically
- * and perform minimum instance check and scaling check using the underlying
- * rules engine.
- */
-abstract public class AbstractMonitor implements Runnable{
-
- private static final Log log = LogFactory.getLog(AbstractMonitor.class);
- // Map<NetworkpartitionId, Network Partition Context>
- protected Map<String, NetworkPartitionContext> networkPartitionCtxts;
- protected DeploymentPolicy deploymentPolicy;
- protected AutoscalePolicy autoscalePolicy;
-
-
- protected FactHandle minCheckFactHandle;
- protected FactHandle scaleCheckFactHandle;
- protected FactHandle terminateDependencyFactHandle;
-
- protected StatefulKnowledgeSession minCheckKnowledgeSession;
- protected StatefulKnowledgeSession scaleCheckKnowledgeSession;
- protected StatefulKnowledgeSession terminateDependencyKnowledgeSession;
- protected boolean isDestroyed;
-
- protected String clusterId;
- protected String serviceId;
-
- protected AutoscalerRuleEvaluator autoscalerRuleEvaluator;
-
- // time intereval between two runs of the Monitor. Default is 90000ms.
- protected int monitorInterval;
-
- public AbstractMonitor() {
- readConfigurations();
- }
-
- private void readConfigurations () {
-
- XMLConfiguration conf = ConfUtil.getInstance(null).getConfiguration();
- monitorInterval = conf.getInt(AutoScalerConstants.AUTOSCALER_MONITOR_INTERVAL, 90000);
- if (log.isDebugEnabled()) {
- log.debug("Cluster Monitor task interval: " + getMonitorInterval());
- }
- }
-
- @Override
- public void run() {
- // TODO Auto-generated method stub
-
- }
-
-
- public NetworkPartitionContext getNetworkPartitionCtxt(Member member) {
- log.info("***** getNetworkPartitionCtxt " + member.getNetworkPartitionId());
- String networkPartitionId = member.getNetworkPartitionId();
- if(networkPartitionCtxts.containsKey(networkPartitionId)) {
- log.info("returnnig network partition context " + networkPartitionCtxts.get(networkPartitionId));
- return networkPartitionCtxts.get(networkPartitionId);
- }
- log.info("returning null getNetworkPartitionCtxt");
- return null;
- }
-
- public String getPartitionOfMember(String memberId){
- for(Service service: TopologyManager.getTopology().getServices()){
- for(Cluster cluster: service.getClusters()){
- if(cluster.memberExists(memberId)){
- return cluster.getMember(memberId).getPartitionId();
- }
- }
- }
- return null;
- }
-
- public void destroy() {
- minCheckKnowledgeSession.dispose();
- scaleCheckKnowledgeSession.dispose();
- terminateDependencyKnowledgeSession.dispose();
- setDestroyed(true);
- if (log.isDebugEnabled()) {
- log.debug("Cluster Monitor Drools session has been disposed. " + this.toString());
- }
- }
-
- public boolean isDestroyed() {
- return isDestroyed;
- }
-
- public void setDestroyed(boolean isDestroyed) {
- this.isDestroyed = isDestroyed;
- }
-
- public String getServiceId() {
- return serviceId;
- }
-
- public void setServiceId(String serviceId) {
- this.serviceId = serviceId;
- }
-
- public DeploymentPolicy getDeploymentPolicy() {
- return deploymentPolicy;
- }
-
- public void setDeploymentPolicy(DeploymentPolicy deploymentPolicy) {
- this.deploymentPolicy = deploymentPolicy;
- }
-
- public AutoscalePolicy getAutoscalePolicy() {
- return autoscalePolicy;
- }
-
- public void setAutoscalePolicy(AutoscalePolicy autoscalePolicy) {
- this.autoscalePolicy = autoscalePolicy;
- }
-
- public String getClusterId() {
- return clusterId;
- }
-
- public void setClusterId(String clusterId) {
- this.clusterId = clusterId;
- }
-
- public Map<String, NetworkPartitionContext> getNetworkPartitionCtxts() {
- return networkPartitionCtxts;
- }
-
- public NetworkPartitionContext getNetworkPartitionCtxt(String networkPartitionId) {
- return networkPartitionCtxts.get(networkPartitionId);
- }
-
- public void setPartitionCtxt(Map<String, NetworkPartitionContext> partitionCtxt) {
- this.networkPartitionCtxts = partitionCtxt;
- }
-
- public boolean partitionCtxtAvailable(String partitionId) {
- return networkPartitionCtxts.containsKey(partitionId);
- }
-
- public void addNetworkPartitionCtxt(NetworkPartitionContext ctxt) {
- this.networkPartitionCtxts.put(ctxt.getId(), ctxt);
- }
-
- public NetworkPartitionContext getPartitionCtxt(String id) {
- return this.networkPartitionCtxts.get(id);
- }
-
- public StatefulKnowledgeSession getMinCheckKnowledgeSession() {
- return minCheckKnowledgeSession;
- }
-
- public void setMinCheckKnowledgeSession(StatefulKnowledgeSession minCheckKnowledgeSession) {
- this.minCheckKnowledgeSession = minCheckKnowledgeSession;
- }
-
- public FactHandle getMinCheckFactHandle() {
- return minCheckFactHandle;
- }
-
- public void setMinCheckFactHandle(FactHandle minCheckFactHandle) {
- this.minCheckFactHandle = minCheckFactHandle;
- }
-
- public StatefulKnowledgeSession getTerminateDependencyKnowledgeSession() {
- return terminateDependencyKnowledgeSession;
- }
-
- public void setTerminateDependencyKnowledgeSession(
- StatefulKnowledgeSession terminateDependencyKnowledgeSession) {
- this.terminateDependencyKnowledgeSession = terminateDependencyKnowledgeSession;
- }
-
- public FactHandle getTerminateDependencyFactHandle() {
- return terminateDependencyFactHandle;
- }
-
- public void setTerminateDependencyFactHandle(
- FactHandle terminateDependencyFactHandle) {
- this.terminateDependencyFactHandle = terminateDependencyFactHandle;
- }
-
- public int getMonitorInterval() {
- return monitorInterval;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/stratos/blob/1422611d/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/ClusterMonitor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/ClusterMonitor.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/ClusterMonitor.java
deleted file mode 100644
index a17bf55..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/ClusterMonitor.java
+++ /dev/null
@@ -1,244 +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.monitor;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.NetworkPartitionContext;
-import org.apache.stratos.autoscaler.PartitionContext;
-import org.apache.stratos.autoscaler.deployment.policy.DeploymentPolicy;
-import org.apache.stratos.autoscaler.policy.model.AutoscalePolicy;
-import org.apache.stratos.autoscaler.rule.AutoscalerRuleEvaluator;
-import org.apache.stratos.cloud.controller.stub.pojo.MemberContext;
-import org.apache.stratos.cloud.controller.stub.pojo.Properties;
-import org.apache.stratos.cloud.controller.stub.pojo.Property;
-import org.apache.stratos.messaging.domain.topology.ClusterStatus;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * Is responsible for monitoring a service cluster. This runs periodically
- * and perform minimum instance check and scaling check using the underlying
- * rules engine.
- */
-public class ClusterMonitor extends AbstractMonitor {
-
- private static final Log log = LogFactory.getLog(ClusterMonitor.class);
- private String lbReferenceType;
- private boolean hasPrimary;
- private ClusterStatus status;
-
- public ClusterMonitor(String clusterId, String serviceId, DeploymentPolicy deploymentPolicy,
- AutoscalePolicy autoscalePolicy) {
- this.clusterId = clusterId;
- this.serviceId = serviceId;
-
- this.autoscalerRuleEvaluator = new AutoscalerRuleEvaluator();
- this.scaleCheckKnowledgeSession = autoscalerRuleEvaluator.getScaleCheckStatefulSession();
- this.minCheckKnowledgeSession = autoscalerRuleEvaluator.getMinCheckStatefulSession();
- this.terminateDependencyKnowledgeSession = autoscalerRuleEvaluator.getTerminateDependencyStatefulSession();
-
- this.deploymentPolicy = deploymentPolicy;
- this.autoscalePolicy = autoscalePolicy;
- if (log.isDebugEnabled()) {
- log.debug("ClusterMonitor:autoScalePolicy:" + autoscalePolicy);
- }
- networkPartitionCtxts = new ConcurrentHashMap<String, NetworkPartitionContext>();
- }
-
-
- @Override
- public void run() {
-
- try {
- // TODO make this configurable,
- // this is the delay the min check of normal cluster monitor to wait until LB monitor is added
- Thread.sleep(60000);
- } catch (InterruptedException ignore) {
- }
-
- while (!isDestroyed()) {
- if (log.isDebugEnabled()) {
- log.debug("Cluster monitor is running.. " + this.toString());
- }
- try {
- if(!ClusterStatus.In_Maintenance.equals(status)) {
- monitor();
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Cluster monitor is suspended as the cluster is in " +
- ClusterStatus.In_Maintenance + " mode......");
- }
- }
- } catch (Exception e) {
- log.error("Cluster monitor: Monitor failed." + this.toString(), e);
- }
- try {
- Thread.sleep(monitorInterval);
- } catch (InterruptedException ignore) {
- }
- }
- }
-
- private boolean isPrimaryMember(MemberContext memberContext){
- Properties props = memberContext.getProperties();
- if (log.isDebugEnabled()) {
- log.debug(" Properties [" + props + "] ");
- }
- if (props != null && props.getProperties() != null) {
- for (Property prop : props.getProperties()) {
- if (prop.getName().equals("PRIMARY")) {
- if (Boolean.parseBoolean(prop.getValue())) {
- log.debug("Adding member id [" + memberContext.getMemberId() + "] " +
- "member instance id [" + memberContext.getInstanceId() + "] as a primary member");
- return true;
- }
- }
- }
- }
- return false;
- }
-
- private void monitor() {
- //TODO make this concurrent
- for (NetworkPartitionContext networkPartitionContext : networkPartitionCtxts.values()) {
- // store primary members in the network partition context
- List<String> primaryMemberListInNetworkPartition = new ArrayList<String>();
-
- //minimum check per partition
- for (PartitionContext partitionContext : networkPartitionContext.getPartitionCtxts().values()) {
- // store primary members in the partition context
- List<String> primaryMemberListInPartition = new ArrayList<String>();
- // get active primary members in this partition context
- for (MemberContext memberContext : partitionContext.getActiveMembers()) {
- if (isPrimaryMember(memberContext)){
- primaryMemberListInPartition.add(memberContext.getMemberId());
- }
- }
- // get pending primary members in this partition context
- for (MemberContext memberContext : partitionContext.getPendingMembers()) {
- if (isPrimaryMember(memberContext)){
- primaryMemberListInPartition.add(memberContext.getMemberId());
- }
- }
- primaryMemberListInNetworkPartition.addAll(primaryMemberListInPartition);
- minCheckKnowledgeSession.setGlobal("clusterId", clusterId);
- minCheckKnowledgeSession.setGlobal("lbRef", lbReferenceType);
- minCheckKnowledgeSession.setGlobal("isPrimary", hasPrimary);
-
-
- if (log.isDebugEnabled()) {
- log.debug(String.format("Running minimum check for partition %s ", partitionContext.getPartitionId()));
- }
-
- minCheckFactHandle = AutoscalerRuleEvaluator.evaluateMinCheck(minCheckKnowledgeSession
- , minCheckFactHandle, partitionContext);
-
- }
-
- //terminate dependency per partition
- // rule terminates all members of a service which is a prerequisites for other services
- for (PartitionContext partitionContext : networkPartitionContext.getPartitionCtxts().values()) {
-
- terminateDependencyKnowledgeSession.setGlobal("clusterId", clusterId);
- terminateDependencyKnowledgeSession.setGlobal("lbRef", lbReferenceType);
- terminateDependencyKnowledgeSession.setGlobal("autoscalePolicy", autoscalePolicy);
- terminateDependencyKnowledgeSession.setGlobal("serviceId", serviceId);
-
- if (log.isDebugEnabled()) {
- log.debug(String.format("Running terminate dependency for partition %s ", partitionContext.getPartitionId()));
- }
-
- terminateDependencyFactHandle = AutoscalerRuleEvaluator.evaluateTerminateDependency(terminateDependencyKnowledgeSession
- , terminateDependencyFactHandle, partitionContext);
-
- }
-
- boolean rifReset = networkPartitionContext.isRifReset();
- boolean memoryConsumptionReset = networkPartitionContext.isMemoryConsumptionReset();
- boolean loadAverageReset = networkPartitionContext.isLoadAverageReset();
-
- if (log.isDebugEnabled()) {
- log.debug("flag of rifReset: " + rifReset + " flag of memoryConsumptionReset" + memoryConsumptionReset
- + " flag of loadAverageReset" + loadAverageReset);
- }
- if (rifReset || memoryConsumptionReset || loadAverageReset) {
-
- scaleCheckKnowledgeSession.setGlobal("clusterId", clusterId);
- //scaleCheckKnowledgeSession.setGlobal("deploymentPolicy", deploymentPolicy);
- scaleCheckKnowledgeSession.setGlobal("autoscalePolicy", autoscalePolicy);
- scaleCheckKnowledgeSession.setGlobal("rifReset", rifReset);
- scaleCheckKnowledgeSession.setGlobal("mcReset", memoryConsumptionReset);
- scaleCheckKnowledgeSession.setGlobal("laReset", loadAverageReset);
- scaleCheckKnowledgeSession.setGlobal("lbRef", lbReferenceType);
- scaleCheckKnowledgeSession.setGlobal("isPrimary", false);
- scaleCheckKnowledgeSession.setGlobal("primaryMembers", primaryMemberListInNetworkPartition);
-
- if (log.isDebugEnabled()) {
- log.debug(String.format("Running scale check for network partition %s ", networkPartitionContext.getId()));
- log.debug(" Primary members : " + primaryMemberListInNetworkPartition);
- }
-
- scaleCheckFactHandle = AutoscalerRuleEvaluator.evaluateScaleCheck(scaleCheckKnowledgeSession
- , scaleCheckFactHandle, networkPartitionContext);
-
- networkPartitionContext.setRifReset(false);
- networkPartitionContext.setMemoryConsumptionReset(false);
- networkPartitionContext.setLoadAverageReset(false);
- } else if (log.isDebugEnabled()) {
- log.debug(String.format("Scale rule will not run since the LB statistics have not received before this " +
- "cycle for network partition %s", networkPartitionContext.getId()));
- }
- }
- }
-
- @Override
- public String toString() {
- return "ClusterMonitor [clusterId=" + clusterId + ", serviceId=" + serviceId +
- ", deploymentPolicy=" + deploymentPolicy + ", autoscalePolicy=" + autoscalePolicy +
- ", lbReferenceType=" + lbReferenceType +
- ", hasPrimary=" + hasPrimary + " ]";
- }
-
- public String getLbReferenceType() {
- return lbReferenceType;
- }
-
- public void setLbReferenceType(String lbReferenceType) {
- this.lbReferenceType = lbReferenceType;
- }
-
- public boolean isHasPrimary() {
- return hasPrimary;
- }
-
- public void setHasPrimary(boolean hasPrimary) {
- this.hasPrimary = hasPrimary;
- }
-
- public ClusterStatus getStatus() {
- return status;
- }
-
- public void setStatus(ClusterStatus status) {
- this.status = status;
- }
-}