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 2015/01/17 14:48:20 UTC
[1/2] stratos git commit: Capturing thread interrupted exception in
message delegators to handle server shutdown behaviour
Repository: stratos
Updated Branches:
refs/heads/master 781b873e9 -> a6034069f
Capturing thread interrupted exception in message delegators to handle server shutdown behaviour
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/ecb068f4
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/ecb068f4
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/ecb068f4
Branch: refs/heads/master
Commit: ecb068f48178aa496b296a30a6beadc01afa8a3d
Parents: 781b873
Author: Imesh Gunaratne <im...@apache.org>
Authored: Sat Jan 17 18:54:37 2015 +0530
Committer: Imesh Gunaratne <im...@apache.org>
Committed: Sat Jan 17 18:54:37 2015 +0530
----------------------------------------------------------------------
.../application/ApplicationsEventMessageDelegator.java | 3 +++
.../cluster/status/ClusterStatusEventMessageDelegator.java | 9 ++++++---
.../domain/mapping/DomainMappingEventMessageDelegator.java | 4 ++--
.../health/stat/HealthStatEventMessageDelegator.java | 3 +++
.../notifier/InstanceNotifierEventMessageDelegator.java | 3 +++
.../status/InstanceStatusEventMessageDelegator.java | 3 +++
.../receiver/tenant/TenantEventMessageDelegator.java | 3 +++
.../receiver/topology/TopologyEventMessageDelegator.java | 6 +++---
8 files changed, 26 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/ecb068f4/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/application/ApplicationsEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/application/ApplicationsEventMessageDelegator.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/application/ApplicationsEventMessageDelegator.java
index 6958235..fa2f6e9 100644
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/application/ApplicationsEventMessageDelegator.java
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/application/ApplicationsEventMessageDelegator.java
@@ -69,6 +69,9 @@ public class ApplicationsEventMessageDelegator implements Runnable {
}
processorChain.process(type, json, ApplicationManager.getApplications());
}
+ } catch (InterruptedException ignore) {
+ log.info("Shutting down application event message delegator...");
+ terminate();
} catch (Exception e) {
log.error("Failed to retrieve application status event message", e);
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/ecb068f4/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/cluster/status/ClusterStatusEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/cluster/status/ClusterStatusEventMessageDelegator.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/cluster/status/ClusterStatusEventMessageDelegator.java
index fb30fb4..5c9c502 100644
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/cluster/status/ClusterStatusEventMessageDelegator.java
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/cluster/status/ClusterStatusEventMessageDelegator.java
@@ -61,16 +61,19 @@ class ClusterStatusEventMessageDelegator implements Runnable {
// Retrieve the actual message
String json = message.getText();
if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster Status event message received from queue: %s", type));
+ log.debug(String.format("Cluster status event message received from queue: %s", type));
}
// Delegate message to message processor chain
if (log.isDebugEnabled()) {
- log.debug(String.format("Delegating Cluster Status event message: %s", type));
+ log.debug(String.format("Delegating cluster status event message: %s", type));
}
processorChain.process(type, json, null);
+ } catch (InterruptedException ignore) {
+ log.info("Shutting down cluster status event message delegator...");
+ terminate();
} catch (Exception e) {
- log.error("Failed to retrieve Cluster Status event message", e);
+ log.error("Failed to retrieve cluster status event message", e);
}
}
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/stratos/blob/ecb068f4/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/domain/mapping/DomainMappingEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/domain/mapping/DomainMappingEventMessageDelegator.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/domain/mapping/DomainMappingEventMessageDelegator.java
index 2af000d..3ebff1a 100644
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/domain/mapping/DomainMappingEventMessageDelegator.java
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/domain/mapping/DomainMappingEventMessageDelegator.java
@@ -71,8 +71,8 @@ class DomainMappingEventMessageDelegator implements Runnable {
}
processorChain.process(type, json, DomainMappingManager.getInstance());
} catch (InterruptedException e) {
- log.info("Domain mapping event message delegator is shutting down...");
- return;
+ log.info("Shutting down domain mapping event message delegator...");
+ terminate();
} catch (Exception e) {
log.error("Failed to retrieve domain mapping event message", e);
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/ecb068f4/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/health/stat/HealthStatEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/health/stat/HealthStatEventMessageDelegator.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/health/stat/HealthStatEventMessageDelegator.java
index 8dcaadd..d75c5fd 100644
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/health/stat/HealthStatEventMessageDelegator.java
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/health/stat/HealthStatEventMessageDelegator.java
@@ -82,6 +82,9 @@ class HealthStatEventMessageDelegator implements Runnable {
log.debug(String.format("Delegating instance notifier event message: %s", type));
}
processorChain.process(type, json, null);
+ } catch (InterruptedException ignore) {
+ log.info("Shutting down health statistics event message delegator...");
+ terminate();
} catch (Exception e) {
log.error("Failed to retrieve instance notifier event message", e);
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/ecb068f4/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/notifier/InstanceNotifierEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/notifier/InstanceNotifierEventMessageDelegator.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/notifier/InstanceNotifierEventMessageDelegator.java
index 538360d..73ef9fe 100644
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/notifier/InstanceNotifierEventMessageDelegator.java
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/notifier/InstanceNotifierEventMessageDelegator.java
@@ -69,6 +69,9 @@ class InstanceNotifierEventMessageDelegator implements Runnable {
log.debug(String.format("Delegating instance notifier event message: %s", type));
}
processorChain.process(type, json, null);
+ } catch (InterruptedException ignore) {
+ log.info("Shutting down instance notifier event message delegator...");
+ terminate();
} catch (Exception e) {
log.error("Failed to retrieve instance notifier event message", e);
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/ecb068f4/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/status/InstanceStatusEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/status/InstanceStatusEventMessageDelegator.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/status/InstanceStatusEventMessageDelegator.java
index cff26c6..9f754b0 100644
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/status/InstanceStatusEventMessageDelegator.java
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/instance/status/InstanceStatusEventMessageDelegator.java
@@ -69,6 +69,9 @@ class InstanceStatusEventMessageDelegator implements Runnable {
log.debug(String.format("Delegating instance notifier event message: %s", type));
}
processorChain.process(type, json, null);
+ } catch (InterruptedException ignore) {
+ log.info("Shutting down instance status event message delegator...");
+ terminate();
} catch (Exception e) {
log.error("Failed to retrieve instance notifier event message", e);
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/ecb068f4/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/tenant/TenantEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/tenant/TenantEventMessageDelegator.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/tenant/TenantEventMessageDelegator.java
index 2b2214b..a295d40 100644
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/tenant/TenantEventMessageDelegator.java
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/tenant/TenantEventMessageDelegator.java
@@ -73,6 +73,9 @@ class TenantEventMessageDelegator implements Runnable {
log.debug(String.format("Delegating tenant event message: %s", type));
}
processorChain.process(type, json, null);
+ } catch (InterruptedException ignore) {
+ log.info("Shutting down tenant event message delegator...");
+ terminate();
} catch (Exception e) {
log.error("Failed to retrieve tenant event message", e);
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/ecb068f4/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/topology/TopologyEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/topology/TopologyEventMessageDelegator.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/topology/TopologyEventMessageDelegator.java
index cd3424c..8508d91 100644
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/topology/TopologyEventMessageDelegator.java
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/receiver/topology/TopologyEventMessageDelegator.java
@@ -71,9 +71,9 @@ class TopologyEventMessageDelegator implements Runnable {
log.debug(String.format("Delegating topology event message: %s", type));
}
processorChain.process(type, json, TopologyManager.getTopology());
- } catch (InterruptedException e) {
- log.info("TopologyEventMessageDelegator is shutting down ..");
- return;
+ } catch (InterruptedException ignore) {
+ log.info("Shutting down topology event message delegator...");
+ terminate();
} catch (Exception e) {
log.error("Failed to retrieve topology event message", e);
}
[2/2] stratos git commit: Fixing executor service shutdown logic in
service components
Posted by im...@apache.org.
Fixing executor service shutdown logic in service components
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/a6034069
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/a6034069
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/a6034069
Branch: refs/heads/master
Commit: a6034069fc7df01a89aa350d952e3ad85fd59589
Parents: ecb068f
Author: Imesh Gunaratne <im...@apache.org>
Authored: Sat Jan 17 19:02:24 2015 +0530
Committer: Imesh Gunaratne <im...@apache.org>
Committed: Sat Jan 17 19:02:24 2015 +0530
----------------------------------------------------------------------
.../internal/AutoscalerServerComponent.java | 254 -----------------
.../internal/AutoscalerServiceComponent.java | 273 +++++++++++++++++++
.../component/ParentComponentMonitor.java | 9 +-
.../autoscaler/util/AutoScalerConstants.java | 11 +-
.../stratos/autoscaler/util/AutoscalerUtil.java | 158 +----------
.../autoscaler/util/ServiceReferenceHolder.java | 13 +-
.../CloudControllerServiceComponent.java | 18 +-
.../TenantAwareLoadBalanceEndpoint.java | 48 ++--
.../internal/LoadBalancerServiceComponent.java | 42 ++-
.../balancer/mediators/LocationReWriter.java | 6 +-
.../balancer/mediators/ResponseInterceptor.java | 4 +-
.../LoadBalancerTopologyEventReceiver.java | 43 +--
.../LoadBalancerStatisticsExecutor.java | 13 +-
.../stratos/load/balancer/util/Constants.java | 38 ---
.../balancer/util/LoadBalancerConstants.java | 40 +++
.../StratosManagerServiceComponent.java | 2 +-
.../StratosManagerTopologyEventReceiver.java | 12 -
.../extension/FaultHandlingWindowProcessor.java | 24 +-
18 files changed, 466 insertions(+), 542 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java
deleted file mode 100644
index f9b743d..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java
+++ /dev/null
@@ -1,254 +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.internal;
-
-import org.apache.commons.configuration.XMLConfiguration;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.applications.ApplicationSynchronizerTaskScheduler;
-import org.apache.stratos.autoscaler.context.AutoscalerContext;
-import org.apache.stratos.autoscaler.event.receiver.health.AutoscalerHealthStatEventReceiver;
-import org.apache.stratos.autoscaler.event.receiver.topology.AutoscalerTopologyEventReceiver;
-import org.apache.stratos.autoscaler.exception.AutoScalerException;
-import org.apache.stratos.autoscaler.pojo.policy.PolicyManager;
-import org.apache.stratos.autoscaler.pojo.policy.autoscale.AutoscalePolicy;
-import org.apache.stratos.autoscaler.pojo.policy.deployment.DeploymentPolicy;
-import org.apache.stratos.autoscaler.registry.RegistryManager;
-import org.apache.stratos.autoscaler.status.processor.cluster.ClusterStatusProcessorChain;
-import org.apache.stratos.autoscaler.status.processor.group.GroupStatusProcessorChain;
-import org.apache.stratos.autoscaler.util.ConfUtil;
-import org.apache.stratos.autoscaler.util.ServiceReferenceHolder;
-import org.apache.stratos.cloud.controller.stub.domain.Partition;
-import org.apache.stratos.common.clustering.DistributedObjectProvider;
-import org.apache.stratos.common.threading.StratosThreadPool;
-import org.osgi.service.component.ComponentContext;
-import org.wso2.carbon.ntask.core.service.TaskService;
-import org.wso2.carbon.registry.api.RegistryException;
-import org.wso2.carbon.registry.core.service.RegistryService;
-import org.wso2.carbon.utils.CarbonUtils;
-import org.wso2.carbon.utils.ConfigurationContextService;
-
-import com.hazelcast.core.HazelcastInstance;
-
-import java.io.File;
-import java.util.Iterator;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-
-/**
- * @scr.component name=org.apache.stratos.autoscaler.internal.AutoscalerServerComponent" immediate="true"
- * @scr.reference name="registry.service" interface="org.wso2.carbon.registry.core.service.RegistryService"
- * cardinality="1..1" policy="dynamic" bind="setRegistryService" unbind="unsetRegistryService"
- * @scr.reference name="ntask.component" interface="org.wso2.carbon.ntask.core.service.TaskService"
- * cardinality="1..1" policy="dynamic" bind="setTaskService" unbind="unsetTaskService"
- * @scr.reference name="hazelcast.instance.service" interface="com.hazelcast.core.HazelcastInstance"
- * cardinality="0..1"policy="dynamic" bind="setHazelcastInstance" unbind="unsetHazelcastInstance"
- * @scr.reference name="distributedObjectProvider" interface="org.apache.stratos.common.clustering.DistributedObjectProvider"
- * cardinality="1..1" policy="dynamic" bind="setDistributedObjectProvider" unbind="unsetDistributedObjectProvider"
- * @scr.reference name="config.context.service" interface="org.wso2.carbon.utils.ConfigurationContextService"
- * cardinality="1..1" policy="dynamic" bind="setConfigurationContextService" unbind="unsetConfigurationContextService"
- */
-
-public class AutoscalerServerComponent {
-
- private static final String THREAD_IDENTIFIER_KEY = "threadPool.autoscaler.identifier";
- private static final String DEFAULT_IDENTIFIER = "Auto-Scaler";
- private static final String THREAD_POOL_SIZE_KEY = "threadPool.autoscaler.threadPoolSize";
- private static final String COMPONENTS_CONFIG = CarbonUtils.getCarbonConfigDirPath()+ File.separator+"stratos-config.xml";
- private static final int THREAD_POOL_SIZE = 10;
- private static final Log log = LogFactory.getLog(AutoscalerServerComponent.class);
- private static final String AUTOSCALER_COORDINATOR_LOCK = "AUTOSCALER_COORDINATOR_LOCK";
-
- private AutoscalerTopologyEventReceiver asTopologyReceiver;
- private AutoscalerHealthStatEventReceiver autoscalerHealthStatEventReceiver;
- private ExecutorService executorService;
-
- protected void activate(ComponentContext componentContext) throws Exception {
- try {
-
- XMLConfiguration conf = ConfUtil.getInstance(COMPONENTS_CONFIG).getConfiguration();
- int threadPoolSize = conf.getInt(THREAD_POOL_SIZE_KEY, THREAD_POOL_SIZE);
- String threadIdentifier = conf.getString(THREAD_IDENTIFIER_KEY, DEFAULT_IDENTIFIER);
- executorService = StratosThreadPool.getExecutorService(threadIdentifier, threadPoolSize);
-
- if(AutoscalerContext.getInstance().isClustered()) {
- Thread coordinatorElectorThread = new Thread() {
- @Override
- public void run() {
- ServiceReferenceHolder.getInstance().getHazelcastInstance()
- .getLock(AUTOSCALER_COORDINATOR_LOCK).lock();
-
- log.info("Elected this member [" + ServiceReferenceHolder.getInstance().getHazelcastInstance()
- .getCluster().getLocalMember().getUuid() + "] " +
- "as the autoscaler coordinator for the cluster");
-
- AutoscalerContext.getInstance().setCoordinator(true);
- try {
- executeCoordinatorTasks();
- } catch (Throwable e) {
- log.error("Error in activating the autoscaler component ", e);
- }
- }
- };
- coordinatorElectorThread.setName("Autoscaler coordinator elector thread");
- executorService.submit(coordinatorElectorThread);
- } else {
- executeCoordinatorTasks();
- }
-
- if (log.isInfoEnabled()) {
- log.info("Autoscaler server Component activated");
- }
- } catch (Throwable e) {
- log.error("Error in activating the autoscaler component ", e);
- }
- }
-
- private void executeCoordinatorTasks() throws Exception{
-
- // Start topology receiver
- asTopologyReceiver = new AutoscalerTopologyEventReceiver();
- asTopologyReceiver.setExecutorService(executorService);
- asTopologyReceiver.execute();
-
- if (log.isDebugEnabled()) {
- log.debug("Topology receiver executor service started");
- }
-
- // Start health stat receiver
- autoscalerHealthStatEventReceiver = new AutoscalerHealthStatEventReceiver();
- autoscalerHealthStatEventReceiver.setExecutorService(executorService);
- autoscalerHealthStatEventReceiver.execute();
- if (log.isDebugEnabled()) {
- log.debug("Health statistics receiver thread started");
- }
-
- // Adding the registry stored partitions to the information model
- List<Partition> partitions = RegistryManager.getInstance().retrievePartitions();
- Iterator<Partition> partitionIterator = partitions.iterator();
- while (partitionIterator.hasNext()) {
- Partition partition = partitionIterator.next();
-// PartitionManager.getInstance().addPartitionToInformationModel(partition);
- }
-
- // Adding the network partitions stored in registry to the information model
-// List<NetworkPartitionLbHolder> nwPartitionHolders = RegistryManager.getInstance().retrieveNetworkPartitionLbHolders();
-// Iterator<NetworkPartitionLbHolder> nwPartitionIterator = nwPartitionHolders.iterator();
-// while (nwPartitionIterator.hasNext()) {
-// NetworkPartitionLbHolder nwPartition = nwPartitionIterator.next();
-// PartitionManager.getInstance().addNetworkPartitionLbHolder(nwPartition);
-// }
-
- // Add AS policies to information model
- List<AutoscalePolicy> asPolicies = RegistryManager.getInstance().retrieveASPolicies();
- Iterator<AutoscalePolicy> asPolicyIterator = asPolicies.iterator();
- while (asPolicyIterator.hasNext()) {
- AutoscalePolicy asPolicy = asPolicyIterator.next();
- PolicyManager.getInstance().addASPolicyToInformationModel(asPolicy);
- }
-
- // Add Deployment policies to information model
- List<DeploymentPolicy> depPolicies = RegistryManager.getInstance().retrieveDeploymentPolicies();
- Iterator<DeploymentPolicy> depPolicyIterator = depPolicies.iterator();
- while (depPolicyIterator.hasNext()) {
- DeploymentPolicy depPolicy = depPolicyIterator.next();
- PolicyManager.getInstance().addDeploymentPolicy(depPolicy);
- }
-
- //starting the processor chain
- ClusterStatusProcessorChain clusterStatusProcessorChain = new ClusterStatusProcessorChain();
- ServiceReferenceHolder.getInstance().setClusterStatusProcessorChain(clusterStatusProcessorChain);
-
- GroupStatusProcessorChain groupStatusProcessorChain = new GroupStatusProcessorChain();
- ServiceReferenceHolder.getInstance().setGroupStatusProcessorChain(groupStatusProcessorChain);
-
- if (log.isInfoEnabled()) {
- log.info("Scheduling tasks to publish applications");
- }
-
- ApplicationSynchronizerTaskScheduler
- .schedule(ServiceReferenceHolder.getInstance()
- .getTaskService());
- }
-
- protected void deactivate(ComponentContext context) {
- asTopologyReceiver.terminate();
- autoscalerHealthStatEventReceiver.terminate();
- }
-
- protected void setRegistryService(RegistryService registryService) {
- if (log.isDebugEnabled()) {
- log.debug("Setting the Registry Service");
- }
- try {
- ServiceReferenceHolder.getInstance().setRegistry(registryService.getGovernanceSystemRegistry());
- } catch (RegistryException e) {
- String msg = "Failed when retrieving Governance System Registry.";
- log.error(msg, e);
- throw new AutoScalerException(msg, e);
- }
- }
-
- protected void unsetRegistryService(RegistryService registryService) {
- if (log.isDebugEnabled()) {
- log.debug("Un-setting the Registry Service");
- }
- ServiceReferenceHolder.getInstance().setRegistry(null);
- }
-
- protected void setTaskService(TaskService taskService) {
- if (log.isDebugEnabled()) {
- log.debug("Setting the Task Service");
- }
- ServiceReferenceHolder.getInstance().setTaskService(taskService);
- }
-
- protected void unsetTaskService(TaskService taskService) {
- if (log.isDebugEnabled()) {
- log.debug("Un-setting the Task Service");
- }
- ServiceReferenceHolder.getInstance().setTaskService(null);
- }
-
- protected void setConfigurationContextService(ConfigurationContextService cfgCtxService) {
- ServiceReferenceHolder.getInstance().setAxisConfiguration(
- cfgCtxService.getServerConfigContext().getAxisConfiguration());
- }
-
- protected void unsetConfigurationContextService(ConfigurationContextService cfgCtxService) {
- ServiceReferenceHolder.getInstance().setAxisConfiguration(null);
- }
-
- public void setHazelcastInstance(HazelcastInstance hazelcastInstance) {
- ServiceReferenceHolder.getInstance().setHazelcastInstance(hazelcastInstance);
- }
-
- public void unsetHazelcastInstance(HazelcastInstance hazelcastInstance) {
- ServiceReferenceHolder.getInstance().setHazelcastInstance(null);
- }
-
- protected void setDistributedObjectProvider(DistributedObjectProvider distributedObjectProvider) {
- ServiceReferenceHolder.getInstance().setDistributedObjectProvider(distributedObjectProvider);
- }
-
- protected void unsetDistributedObjectProvider(DistributedObjectProvider distributedObjectProvider) {
- ServiceReferenceHolder.getInstance().setDistributedObjectProvider(null);
- }
-}
-
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServiceComponent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServiceComponent.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServiceComponent.java
new file mode 100644
index 0000000..5373850
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServiceComponent.java
@@ -0,0 +1,273 @@
+/*
+ * 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.internal;
+
+import org.apache.commons.configuration.XMLConfiguration;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.autoscaler.applications.ApplicationSynchronizerTaskScheduler;
+import org.apache.stratos.autoscaler.context.AutoscalerContext;
+import org.apache.stratos.autoscaler.event.receiver.health.AutoscalerHealthStatEventReceiver;
+import org.apache.stratos.autoscaler.event.receiver.topology.AutoscalerTopologyEventReceiver;
+import org.apache.stratos.autoscaler.exception.AutoScalerException;
+import org.apache.stratos.autoscaler.exception.policy.InvalidPolicyException;
+import org.apache.stratos.autoscaler.pojo.policy.PolicyManager;
+import org.apache.stratos.autoscaler.pojo.policy.autoscale.AutoscalePolicy;
+import org.apache.stratos.autoscaler.pojo.policy.deployment.DeploymentPolicy;
+import org.apache.stratos.autoscaler.registry.RegistryManager;
+import org.apache.stratos.autoscaler.status.processor.cluster.ClusterStatusProcessorChain;
+import org.apache.stratos.autoscaler.status.processor.group.GroupStatusProcessorChain;
+import org.apache.stratos.autoscaler.util.AutoScalerConstants;
+import org.apache.stratos.autoscaler.util.ConfUtil;
+import org.apache.stratos.autoscaler.util.ServiceReferenceHolder;
+import org.apache.stratos.cloud.controller.stub.domain.Partition;
+import org.apache.stratos.common.clustering.DistributedObjectProvider;
+import org.apache.stratos.common.threading.StratosThreadPool;
+import org.osgi.service.component.ComponentContext;
+import org.wso2.carbon.ntask.core.service.TaskService;
+import org.wso2.carbon.registry.api.RegistryException;
+import org.wso2.carbon.registry.core.service.RegistryService;
+import org.wso2.carbon.utils.CarbonUtils;
+import org.wso2.carbon.utils.ConfigurationContextService;
+
+import com.hazelcast.core.HazelcastInstance;
+
+import java.io.File;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * @scr.component name=org.apache.stratos.autoscaler.internal.AutoscalerServiceComponent" immediate="true"
+ * @scr.reference name="registry.service" interface="org.wso2.carbon.registry.core.service.RegistryService"
+ * cardinality="1..1" policy="dynamic" bind="setRegistryService" unbind="unsetRegistryService"
+ * @scr.reference name="ntask.component" interface="org.wso2.carbon.ntask.core.service.TaskService"
+ * cardinality="1..1" policy="dynamic" bind="setTaskService" unbind="unsetTaskService"
+ * @scr.reference name="hazelcast.instance.service" interface="com.hazelcast.core.HazelcastInstance"
+ * cardinality="0..1"policy="dynamic" bind="setHazelcastInstance" unbind="unsetHazelcastInstance"
+ * @scr.reference name="distributedObjectProvider" interface="org.apache.stratos.common.clustering.DistributedObjectProvider"
+ * cardinality="1..1" policy="dynamic" bind="setDistributedObjectProvider" unbind="unsetDistributedObjectProvider"
+ * @scr.reference name="config.context.service" interface="org.wso2.carbon.utils.ConfigurationContextService"
+ * cardinality="1..1" policy="dynamic" bind="setConfigurationContextService" unbind="unsetConfigurationContextService"
+ */
+
+public class AutoscalerServiceComponent {
+
+ private static final Log log = LogFactory.getLog(AutoscalerServiceComponent.class);
+
+ private static final String AUTOSCALER_COORDINATOR_LOCK = "AUTOSCALER_COORDINATOR_LOCK";
+
+ private AutoscalerTopologyEventReceiver asTopologyReceiver;
+ private AutoscalerHealthStatEventReceiver autoscalerHealthStatEventReceiver;
+ private ExecutorService executorService;
+
+ protected void activate(ComponentContext componentContext) throws Exception {
+ try {
+
+ XMLConfiguration conf = ConfUtil.getInstance(AutoScalerConstants.COMPONENTS_CONFIG).getConfiguration();
+ int threadPoolSize = conf.getInt(AutoScalerConstants.THREAD_POOL_SIZE_KEY,
+ AutoScalerConstants.AUTOSCALER_THREAD_POOL_SIZE);
+ executorService = StratosThreadPool.getExecutorService(AutoScalerConstants.AUTOSCALER_THREAD_POOL_ID,
+ threadPoolSize);
+
+ ServiceReferenceHolder.getInstance().setExecutorService(executorService);
+
+ if(AutoscalerContext.getInstance().isClustered()) {
+ Thread coordinatorElectorThread = new Thread() {
+ @Override
+ public void run() {
+ ServiceReferenceHolder.getInstance().getHazelcastInstance()
+ .getLock(AUTOSCALER_COORDINATOR_LOCK).lock();
+
+ log.info("Elected this member [" + ServiceReferenceHolder.getInstance().getHazelcastInstance()
+ .getCluster().getLocalMember().getUuid() + "] " +
+ "as the autoscaler coordinator for the cluster");
+
+ AutoscalerContext.getInstance().setCoordinator(true);
+ try {
+ executeCoordinatorTasks();
+ } catch (Exception e) {
+ log.error("Error in activating the autoscaler component ", e);
+ }
+ }
+ };
+ coordinatorElectorThread.setName("Autoscaler coordinator elector thread");
+ executorService.submit(coordinatorElectorThread);
+ } else {
+ executeCoordinatorTasks();
+ }
+
+ if (log.isInfoEnabled()) {
+ log.info("Autoscaler service component activated");
+ }
+ } catch (Exception e) {
+ log.error("Error in activating the autoscaler service component ", e);
+ }
+ }
+
+ private void executeCoordinatorTasks() throws InvalidPolicyException {
+
+ // Start topology receiver
+ asTopologyReceiver = new AutoscalerTopologyEventReceiver();
+ asTopologyReceiver.setExecutorService(executorService);
+ asTopologyReceiver.execute();
+
+ if (log.isDebugEnabled()) {
+ log.debug("Topology receiver executor service started");
+ }
+
+ // Start health stat receiver
+ autoscalerHealthStatEventReceiver = new AutoscalerHealthStatEventReceiver();
+ autoscalerHealthStatEventReceiver.setExecutorService(executorService);
+ autoscalerHealthStatEventReceiver.execute();
+ if (log.isDebugEnabled()) {
+ log.debug("Health statistics receiver thread started");
+ }
+
+ // Adding the registry stored partitions to the information model
+ List<Partition> partitions = RegistryManager.getInstance().retrievePartitions();
+ Iterator<Partition> partitionIterator = partitions.iterator();
+ while (partitionIterator.hasNext()) {
+ Partition partition = partitionIterator.next();
+// PartitionManager.getInstance().addPartitionToInformationModel(partition);
+ }
+
+ // Adding the network partitions stored in registry to the information model
+// List<NetworkPartitionLbHolder> nwPartitionHolders = RegistryManager.getInstance().retrieveNetworkPartitionLbHolders();
+// Iterator<NetworkPartitionLbHolder> nwPartitionIterator = nwPartitionHolders.iterator();
+// while (nwPartitionIterator.hasNext()) {
+// NetworkPartitionLbHolder nwPartition = nwPartitionIterator.next();
+// PartitionManager.getInstance().addNetworkPartitionLbHolder(nwPartition);
+// }
+
+ // Add AS policies to information model
+ List<AutoscalePolicy> asPolicies = RegistryManager.getInstance().retrieveASPolicies();
+ Iterator<AutoscalePolicy> asPolicyIterator = asPolicies.iterator();
+ while (asPolicyIterator.hasNext()) {
+ AutoscalePolicy asPolicy = asPolicyIterator.next();
+ PolicyManager.getInstance().addASPolicyToInformationModel(asPolicy);
+ }
+
+ // Add Deployment policies to information model
+ List<DeploymentPolicy> depPolicies = RegistryManager.getInstance().retrieveDeploymentPolicies();
+ Iterator<DeploymentPolicy> depPolicyIterator = depPolicies.iterator();
+ while (depPolicyIterator.hasNext()) {
+ DeploymentPolicy depPolicy = depPolicyIterator.next();
+ PolicyManager.getInstance().addDeploymentPolicy(depPolicy);
+ }
+
+ //starting the processor chain
+ ClusterStatusProcessorChain clusterStatusProcessorChain = new ClusterStatusProcessorChain();
+ ServiceReferenceHolder.getInstance().setClusterStatusProcessorChain(clusterStatusProcessorChain);
+
+ GroupStatusProcessorChain groupStatusProcessorChain = new GroupStatusProcessorChain();
+ ServiceReferenceHolder.getInstance().setGroupStatusProcessorChain(groupStatusProcessorChain);
+
+ if (log.isInfoEnabled()) {
+ log.info("Scheduling tasks to publish applications");
+ }
+
+ ApplicationSynchronizerTaskScheduler
+ .schedule(ServiceReferenceHolder.getInstance()
+ .getTaskService());
+ }
+
+ protected void deactivate(ComponentContext context) {
+ try {
+ asTopologyReceiver.terminate();
+ } catch (Exception e) {
+ log.warn("An error occurred while terminating autoscaler topology event receiver", e);
+ }
+
+ try {
+ autoscalerHealthStatEventReceiver.terminate();
+ } catch (Exception e) {
+ log.warn("An error occurred while terminating autoscaler health statistics event receiver", e);
+ }
+
+ // Shutdown executor service
+ if(executorService != null) {
+ try {
+ executorService.shutdownNow();
+ } catch (Exception e) {
+ log.warn("An error occurred while shutting down autoscaler executor service", e);
+ }
+ }
+ }
+
+ protected void setRegistryService(RegistryService registryService) {
+ if (log.isDebugEnabled()) {
+ log.debug("Setting the Registry Service");
+ }
+ try {
+ ServiceReferenceHolder.getInstance().setRegistry(registryService.getGovernanceSystemRegistry());
+ } catch (RegistryException e) {
+ String msg = "Failed when retrieving Governance System Registry.";
+ log.error(msg, e);
+ throw new AutoScalerException(msg, e);
+ }
+ }
+
+ protected void unsetRegistryService(RegistryService registryService) {
+ if (log.isDebugEnabled()) {
+ log.debug("Un-setting the Registry Service");
+ }
+ ServiceReferenceHolder.getInstance().setRegistry(null);
+ }
+
+ protected void setTaskService(TaskService taskService) {
+ if (log.isDebugEnabled()) {
+ log.debug("Setting the Task Service");
+ }
+ ServiceReferenceHolder.getInstance().setTaskService(taskService);
+ }
+
+ protected void unsetTaskService(TaskService taskService) {
+ if (log.isDebugEnabled()) {
+ log.debug("Un-setting the Task Service");
+ }
+ ServiceReferenceHolder.getInstance().setTaskService(null);
+ }
+
+ protected void setConfigurationContextService(ConfigurationContextService cfgCtxService) {
+ ServiceReferenceHolder.getInstance().setAxisConfiguration(
+ cfgCtxService.getServerConfigContext().getAxisConfiguration());
+ }
+
+ protected void unsetConfigurationContextService(ConfigurationContextService cfgCtxService) {
+ ServiceReferenceHolder.getInstance().setAxisConfiguration(null);
+ }
+
+ public void setHazelcastInstance(HazelcastInstance hazelcastInstance) {
+ ServiceReferenceHolder.getInstance().setHazelcastInstance(hazelcastInstance);
+ }
+
+ public void unsetHazelcastInstance(HazelcastInstance hazelcastInstance) {
+ ServiceReferenceHolder.getInstance().setHazelcastInstance(null);
+ }
+
+ protected void setDistributedObjectProvider(DistributedObjectProvider distributedObjectProvider) {
+ ServiceReferenceHolder.getInstance().setDistributedObjectProvider(distributedObjectProvider);
+ }
+
+ protected void unsetDistributedObjectProvider(DistributedObjectProvider distributedObjectProvider) {
+ ServiceReferenceHolder.getInstance().setDistributedObjectProvider(null);
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/component/ParentComponentMonitor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/component/ParentComponentMonitor.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/component/ParentComponentMonitor.java
index d434f83..8531ac6 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/component/ParentComponentMonitor.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/monitor/component/ParentComponentMonitor.java
@@ -43,6 +43,7 @@ import org.apache.stratos.autoscaler.monitor.cluster.ClusterMonitor;
import org.apache.stratos.autoscaler.monitor.events.ScalingDownBeyondMinEvent;
import org.apache.stratos.autoscaler.monitor.events.ScalingEvent;
import org.apache.stratos.autoscaler.monitor.events.ScalingUpBeyondMaxEvent;
+import org.apache.stratos.autoscaler.util.AutoScalerConstants;
import org.apache.stratos.autoscaler.util.ServiceReferenceHolder;
import org.apache.stratos.common.threading.StratosThreadPool;
import org.apache.stratos.messaging.domain.application.GroupStatus;
@@ -60,9 +61,9 @@ import java.util.concurrent.*;
* control them according to the dependencies respectively.
*/
public abstract class ParentComponentMonitor extends Monitor implements Runnable {
+
private static final Log log = LogFactory.getLog(ParentComponentMonitor.class);
- private static final String IDENTIFIER = "Auto-Scaler";
- private static final int THREAD_POOL_SIZE = 10;
+
//Scheduler executor service to execute this monitor in a thread
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
//The monitors dependency tree with all the start-able/kill-able dependencies
@@ -102,9 +103,9 @@ public abstract class ParentComponentMonitor extends Monitor implements Runnable
scalingDependencies = new HashSet<ScalingDependentList>();
}
//Create the executor service with identifier and thread pool size
- executorService = StratosThreadPool.getExecutorService(IDENTIFIER, THREAD_POOL_SIZE);
+ executorService = StratosThreadPool.getExecutorService(AutoScalerConstants.AUTOSCALER_THREAD_POOL_ID,
+ AutoScalerConstants.AUTOSCALER_THREAD_POOL_SIZE);
networkPartitionCtxts = new HashMap<String, NetworkPartitionContext>();
-
}
/**
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoScalerConstants.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoScalerConstants.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoScalerConstants.java
index be86836..7579a01 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoScalerConstants.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoScalerConstants.java
@@ -18,10 +18,14 @@
*/
package org.apache.stratos.autoscaler.util;
+import org.wso2.carbon.utils.CarbonUtils;
+
+import java.io.File;
+
public final class AutoScalerConstants {
/**
- * Constant values for Auto Scaler
+ * Constant values for Autoscaler
*/
public static final String ID_ELEMENT = "id";
public static final String PARTITION_ELEMENT = "partition";
@@ -29,6 +33,11 @@ public final class AutoScalerConstants {
public static final String PROPERTY_ELEMENT = "property";
public static final String PROPERTY_NAME_ATTR= "name";
public static final String PROPERTY_VALUE_ATTR = "value";
+ public static final String AUTOSCALER_THREAD_POOL_ID = "autoscaler.thread.pool";
+ public static final String THREAD_POOL_SIZE_KEY = "autoscaler.thread.pool.size";
+ public static final int AUTOSCALER_THREAD_POOL_SIZE = 10;
+ public static final String COMPONENTS_CONFIG = CarbonUtils.getCarbonConfigDirPath() +
+ File.separator + "stratos-config.xml";
/**
* Persistence
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoscalerUtil.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoscalerUtil.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoscalerUtil.java
index 2921489..61e8fdd 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoscalerUtil.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/AutoscalerUtil.java
@@ -19,13 +19,6 @@
package org.apache.stratos.autoscaler.util;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.concurrent.ExecutorService;
-
-import javax.xml.namespace.QName;
-
import org.apache.axiom.om.OMElement;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
@@ -39,7 +32,6 @@ import org.apache.stratos.autoscaler.monitor.component.ApplicationMonitor;
import org.apache.stratos.autoscaler.registry.RegistryManager;
import org.apache.stratos.common.Properties;
import org.apache.stratos.common.Property;
-import org.apache.stratos.common.threading.StratosThreadPool;
import org.apache.stratos.messaging.domain.application.Application;
import org.apache.stratos.messaging.domain.application.Applications;
import org.apache.stratos.messaging.domain.application.ClusterDataHolder;
@@ -47,6 +39,11 @@ import org.apache.stratos.messaging.domain.topology.Service;
import org.apache.stratos.messaging.domain.topology.Topology;
import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
+import javax.xml.namespace.QName;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Map;
+
/**
* This class contains utility methods used by Autoscaler.
@@ -54,26 +51,14 @@ import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
public class AutoscalerUtil {
private static final Log log = LogFactory.getLog(AutoscalerUtil.class);
- public static final String IDENTIFIER = "Auto-Scaler";
- public static final int THREAD_POOL_SIZE = 10;
- private ExecutorService executorService= StratosThreadPool.getExecutorService(IDENTIFIER, THREAD_POOL_SIZE);
private AutoscalerUtil() {
-
}
public static AutoscalerUtil getInstance() {
return Holder.INSTANCE;
}
- public ExecutorService getExecutorService() {
- return executorService;
- }
-
- public void setExecutorService(ExecutorService executorService) {
- this.executorService = executorService;
- }
-
private static class Holder {
private static final AutoscalerUtil INSTANCE = new AutoscalerUtil();
}
@@ -95,8 +80,8 @@ public class AutoscalerUtil {
}
public static Application getApplication (String appId) {
- return getApplicationFromPath(AutoScalerConstants.AUTOSCALER_RESOURCE + AutoScalerConstants.APPLICATIONS_RESOURCE +
- "/" + appId);
+ return getApplicationFromPath(AutoScalerConstants.AUTOSCALER_RESOURCE +
+ AutoScalerConstants.APPLICATIONS_RESOURCE + "/" + appId);
}
public static void persistApplication (Application application) {
@@ -154,128 +139,6 @@ public class AutoscalerUtil {
return allClustersInitialized;
}
-
-
- /*public static LbClusterMonitor getLBClusterMonitor(Cluster cluster) throws PolicyValidationException, PartitionValidationException {
- // FIXME fix the following code to correctly update
- // AutoscalerContext context = AutoscalerContext.getInstance();
- if (null == cluster) {
- return null;
- }
-
- String autoscalePolicyName = cluster.getAutoscalePolicyName();
- String deploymentPolicyName = cluster.getDeploymentPolicyName();
-
- if (log.isDebugEnabled()) {
- log.debug("Deployment policy name: " + deploymentPolicyName);
- log.debug("Autoscaler policy name: " + autoscalePolicyName);
- }
-
- AutoscalePolicy policy =
- PolicyManager.getInstance()
- .getAutoscalePolicy(autoscalePolicyName);
- DeploymentPolicy deploymentPolicy =
- PolicyManager.getInstance()
- .getDeploymentPolicy(deploymentPolicyName);
-
- if (deploymentPolicy == null) {
- String msg = "Deployment Policy is null. Policy name: " + deploymentPolicyName;
- log.error(msg);
- throw new PolicyValidationException(msg);
- }
-
- String clusterId = cluster.getClusterId();
- LbClusterMonitor clusterMonitor =
- new LbClusterMonitor(clusterId,
- cluster.getServiceName(),
- deploymentPolicy, policy);
- clusterMonitor.notifyParentMonitor(Status.Created);
- // partition group = network partition context
- for (NetworkPartition partitionGroup : deploymentPolicy.gNetworkPartitionups()) {
-
- NetworkPartitionLbHolder networkPartitionLbHolder =
- PartitionManager.getInstance()
- .getNetworkPartitionLbHolder(partitionGroup.getPartitionId());
-// PartitionManager.getInstance()
-// .getNetworkPartitionLbHolder(partitionGroup.getPartitionId());
- // FIXME pick a random partition
- Partition partition =
- partitionGroup.getPartitions()[new Random().nextInt(partitionGroup.getPartitions().length)];
- PartitionContext partitionContext = new PartitionContext(partition);
- partitionContext.setServiceName(cluster.getServiceName());
- partitionContext.setProperties(cluster.getProperties());
- partitionContext.setNetworkPartitionId(partitionGroup.getPartitionId());
- partitionContext.setMinimumMemberCount(1);//Here it hard codes the minimum value as one for LB cartridge partitions
-
- NetworkPartitionContext networkPartitionContext = new NetworkPartitionContext(partitionGroup.getPartitionId(),
- partitionGroup.getPartitionAlgo(), partitionGroup.getPartitions()) ;
- for (Member member : cluster.getMembers()) {
- String memberId = member.getMemberId();
- if (member.getNetworkPartitionId().equalsIgnoreCase(networkPartitionContext.getPartitionId())) {
- MemberContext memberContext = new MemberContext();
- memberContext.setClusterId(member.getClusterId());
- memberContext.setMemberId(memberId);
- memberContext.setPartition(partition);
-
- if (MemberStatus.Activated.equals(member.getStatus())) {
- partitionContext.addActiveMember(memberContext);
-// networkPartitionContext.increaseMemberCountOfPartition(partition.getNetworkPartitionId(), 1);
-// partitionContext.incrementCurrentActiveMemberCount(1);
- } else if (MemberStatus.Created.equals(member.getStatus()) ||
- MemberStatus.Starting.equals(member.getStatus())) {
- partitionContext.addPendingMember(memberContext);
-// networkPartitionContext.increaseMemberCountOfPartition(partition.getNetworkPartitionId(), 1);
- } else if (MemberStatus.Suspended.equals(member.getStatus())) {
-// partitionContext.addFaultyMember(memberId);
- }
-
- partitionContext.addMemberStatsContext(new MemberStatsContext(memberId));
- if(log.isInfoEnabled()){
- log.info(String.format("Member stat context has been added: [member] %s", memberId));
- }
- }
-
- }
- networkPartitionContext.addPartitionContext(partitionContext);
-
- // populate lb cluster id in network partition context.
- java.util.Properties props = cluster.getProperties();
-
- // get service type of load balanced cluster
- String loadBalancedServiceType = props.getProperty(Constants.LOAD_BALANCED_SERVICE_TYPE);
-
- if(props.containsKey(Constants.LOAD_BALANCER_REF)) {
- String value = props.getProperty(Constants.LOAD_BALANCER_REF);
-
- if (value.equals(org.apache.stratos.messaging.util.Constants.DEFAULT_LOAD_BALANCER)) {
- networkPartitionLbHolder.setDefaultLbClusterId(clusterId);
-
- } else if (value.equals(org.apache.stratos.messaging.util.Constants.SERVICE_AWARE_LOAD_BALANCER)) {
- String serviceName = cluster.getServiceName();
- // TODO: check if this is correct
- networkPartitionLbHolder.addServiceLB(serviceName, clusterId);
-
- if (loadBalancedServiceType != null && !loadBalancedServiceType.isEmpty()) {
- networkPartitionLbHolder.addServiceLB(loadBalancedServiceType, clusterId);
- if (log.isDebugEnabled()) {
- log.debug("Added cluster id " + clusterId + " as the LB cluster id for service type " + loadBalancedServiceType);
- }
- }
- }
- }
-
- clusterMonitor.addNetworkPartitionCtxt(networkPartitionContext);
- }
-
- log.info("LB Cluster monitor created: "+clusterMonitor.toString());
- return clusterMonitor;
- }*/
-
- //TODO moving it into factory class
-
-
-
-
public static Properties getProperties(final OMElement elt) {
Iterator<?> it = elt.getChildrenWithName(new QName(AutoScalerConstants.PROPERTY_ELEMENT));
@@ -382,12 +245,11 @@ public class AutoscalerUtil {
AutoscalerContext autoscalerContext = AutoscalerContext.getInstance();
if (autoscalerContext.getAppMonitor(applicationId) == null) {
autoscalerContext.addPendingMonitor(applicationId);
- executorService.submit(new ApplicationMonitorAdder(applicationId));
+ ServiceReferenceHolder.getInstance().getExecutorService().submit(new ApplicationMonitorAdder(applicationId));
} else {
if (log.isDebugEnabled()) {
- log.debug(String
- .format("Application monitor thread already exists: " +
- "[application] %s ", applicationId));
+ log.debug(String.format("Application monitor thread already exists: " +
+ "[application] %s ", applicationId));
}
}
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/ServiceReferenceHolder.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/ServiceReferenceHolder.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/ServiceReferenceHolder.java
index 9930300..9d1fa9c 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/ServiceReferenceHolder.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/util/ServiceReferenceHolder.java
@@ -31,6 +31,8 @@ import org.wso2.carbon.registry.core.session.UserRegistry;
import com.hazelcast.core.HazelcastInstance;
+import java.util.concurrent.ExecutorService;
+
public class ServiceReferenceHolder {
private static ServiceReferenceHolder instance;
@@ -41,8 +43,9 @@ public class ServiceReferenceHolder {
private AxisConfiguration axisConfiguration;
private DistributedObjectProvider distributedObjectProvider;
private HazelcastInstance hazelcastInstance;
+ private ExecutorService executorService;
- private ServiceReferenceHolder() {
+ private ServiceReferenceHolder() {
}
public static ServiceReferenceHolder getInstance() {
@@ -107,4 +110,12 @@ public class ServiceReferenceHolder {
public void setGroupStatusProcessorChain(GroupStatusProcessorChain groupStatusProcessorChain) {
this.groupStatusProcessorChain = groupStatusProcessorChain;
}
+
+ public void setExecutorService(ExecutorService executorService) {
+ this.executorService = executorService;
+ }
+
+ public ExecutorService getExecutorService() {
+ return executorService;
+ }
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/internal/CloudControllerServiceComponent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/internal/CloudControllerServiceComponent.java b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/internal/CloudControllerServiceComponent.java
index bb6fe71..9269ab3 100644
--- a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/internal/CloudControllerServiceComponent.java
+++ b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/internal/CloudControllerServiceComponent.java
@@ -71,10 +71,7 @@ public class CloudControllerServiceComponent {
private ApplicationEventReceiver applicationEventReceiver;
private ExecutorService executorService;
- private static final String THREAD_IDENTIFIER_KEY = "threadPool.autoscaler.identifier";
private static final String DEFAULT_IDENTIFIER = "Cloud-Controller";
- private static final String THREAD_POOL_SIZE_KEY = "threadPool.cloudcontroller.threadPoolSize";
- private static final String COMPONENTS_CONFIG = "stratos-config";
private static final int THREAD_POOL_SIZE = 10;
private static final String CLOUD_CONTROLLER_COORDINATOR_LOCK = "CLOUD_CONTROLLER_COORDINATOR_LOCK";
@@ -212,6 +209,19 @@ public class CloudControllerServiceComponent {
protected void deactivate(ComponentContext ctx) {
// Close event publisher connections to message broker
- EventPublisherPool.close(Util.Topics.TOPOLOGY_TOPIC.getTopicName());
+ try {
+ EventPublisherPool.close(Util.Topics.TOPOLOGY_TOPIC.getTopicName());
+ } catch (Exception e) {
+ log.warn("An error occurred while closing cloud controller topology event publisher", e);
+ }
+
+ // Shutdown executor service
+ if(executorService != null) {
+ try {
+ executorService.shutdownNow();
+ } catch (Exception e) {
+ log.warn("An error occurred while shutting down cloud controller executor service", e);
+ }
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java
index 5aed58c..4c57185 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java
@@ -31,7 +31,7 @@ import org.apache.stratos.load.balancer.context.LoadBalancerContext;
import org.apache.stratos.load.balancer.statistics.InFlightRequestDecrementCallable;
import org.apache.stratos.load.balancer.statistics.InFlightRequestIncrementCallable;
import org.apache.stratos.load.balancer.statistics.LoadBalancerStatisticsExecutor;
-import org.apache.stratos.load.balancer.util.Constants;
+import org.apache.stratos.load.balancer.util.LoadBalancerConstants;
import org.apache.stratos.messaging.domain.tenant.Tenant;
import org.apache.stratos.messaging.domain.topology.Cluster;
import org.apache.stratos.messaging.domain.topology.Member;
@@ -169,12 +169,12 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
TransportInDescription httpsTransportIn = axis2MsgCtx.getConfigurationContext().getAxisConfiguration().getTransportIn(httpsTransportName);
String lbHttpPort = (String) httpTransportIn.getParameter("port").getValue();
String lbHttpsPort = (String) httpsTransportIn.getParameter("port").getValue();
- String clusterId = currentMember.getProperties().getProperty(Constants.CLUSTER_ID);
+ String clusterId = currentMember.getProperties().getProperty(LoadBalancerConstants.CLUSTER_ID);
- synCtx.setProperty(Constants.LB_TARGET_HOSTNAME, targetHostname);
- synCtx.setProperty(Constants.LB_HTTP_PORT, lbHttpPort);
- synCtx.setProperty(Constants.LB_HTTPS_PORT, lbHttpsPort);
- synCtx.setProperty(Constants.CLUSTER_ID, clusterId);
+ synCtx.setProperty(LoadBalancerConstants.LB_TARGET_HOSTNAME, targetHostname);
+ synCtx.setProperty(LoadBalancerConstants.LB_HTTP_PORT, lbHttpPort);
+ synCtx.setProperty(LoadBalancerConstants.LB_HTTPS_PORT, lbHttpsPort);
+ synCtx.setProperty(LoadBalancerConstants.CLUSTER_ID, clusterId);
}
protected String getTransportId(String incomingTransportName) {
@@ -263,8 +263,8 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
axis2Member.setDomain(member.getClusterId());
axis2Member.setActive(member.isActive());
// Set cluster id and member id in member properties
- axis2Member.getProperties().setProperty(Constants.CLUSTER_ID, member.getClusterId());
- axis2Member.getProperties().setProperty(Constants.MEMBER_ID, member.getMemberId());
+ axis2Member.getProperties().setProperty(LoadBalancerConstants.CLUSTER_ID, member.getClusterId());
+ axis2Member.getProperties().setProperty(LoadBalancerConstants.MEMBER_ID, member.getMemberId());
// Update axis2 member ports
updateAxis2MemberPorts(synCtx, axis2Member);
return axis2Member;
@@ -288,13 +288,13 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
if (outgoingPort == null) {
if (log.isErrorEnabled()) {
log.error(String.format("Could not find the port for proxy port %d in member %s", incomingPort,
- axis2Member.getProperties().getProperty(Constants.MEMBER_ID)));
+ axis2Member.getProperties().getProperty(LoadBalancerConstants.MEMBER_ID)));
}
throwSynapseException(synCtx, 500, "Internal server error");
}
- if (Constants.HTTP.equals(transport)) {
+ if (LoadBalancerConstants.HTTP.equals(transport)) {
axis2Member.setHttpPort(outgoingPort.getValue());
- } else if (Constants.HTTPS.equals(transport)) {
+ } else if (LoadBalancerConstants.HTTPS.equals(transport)) {
axis2Member.setHttpsPort(outgoingPort.getValue());
}
}
@@ -310,10 +310,10 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
org.apache.axis2.context.MessageContext msgCtx =
((Axis2MessageContext) synCtx).getAxis2MessageContext();
try {
- String servicePrefix = (String) msgCtx.getProperty(Constants.AXIS2_MSG_CTX_SERVICE_PREFIX);
+ String servicePrefix = (String) msgCtx.getProperty(LoadBalancerConstants.AXIS2_MSG_CTX_SERVICE_PREFIX);
if (servicePrefix == null) {
if (log.isErrorEnabled()) {
- log.error(String.format("%s property not found in axis2 message context", Constants.AXIS2_MSG_CTX_SERVICE_PREFIX));
+ log.error(String.format("%s property not found in axis2 message context", LoadBalancerConstants.AXIS2_MSG_CTX_SERVICE_PREFIX));
}
throwSynapseException(synCtx, 500, "Internal server error");
}
@@ -370,8 +370,8 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
* @return
*/
private Member findMemberFromAxis2Member(MessageContext synCtx, org.apache.axis2.clustering.Member axis2Member) {
- String clusterId = axis2Member.getProperties().getProperty(Constants.CLUSTER_ID);
- String memberId = axis2Member.getProperties().getProperty(Constants.MEMBER_ID);
+ String clusterId = axis2Member.getProperties().getProperty(LoadBalancerConstants.CLUSTER_ID);
+ String memberId = axis2Member.getProperties().getProperty(LoadBalancerConstants.MEMBER_ID);
if (StringUtils.isBlank(clusterId) || StringUtils.isBlank(memberId)) {
if (log.isErrorEnabled()) {
log.error(String.format("Could not find cluster id and/or member id properties in axis2 member: [cluster-id] %s " +
@@ -446,7 +446,7 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
private String extractUrl(MessageContext synCtx) {
Axis2MessageContext axis2smc = (Axis2MessageContext) synCtx;
org.apache.axis2.context.MessageContext axis2MessageCtx = axis2smc.getAxis2MessageContext();
- return (String) axis2MessageCtx.getProperty(Constants.AXIS2_MSG_CTX_TRANSPORT_IN_URL);
+ return (String) axis2MessageCtx.getProperty(LoadBalancerConstants.AXIS2_MSG_CTX_TRANSPORT_IN_URL);
}
/**
@@ -571,10 +571,10 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
private EndpointReference getEndpointReferenceAfterURLRewrite(MessageContext synCtx, org.apache.axis2.clustering.Member currentMember,
String transport) {
try {
- if (transport.startsWith(Constants.HTTPS)) {
- transport = Constants.HTTPS;
- } else if (transport.startsWith(Constants.HTTP)) {
- transport = Constants.HTTP;
+ if (transport.startsWith(LoadBalancerConstants.HTTPS)) {
+ transport = LoadBalancerConstants.HTTPS;
+ } else if (transport.startsWith(LoadBalancerConstants.HTTP)) {
+ transport = LoadBalancerConstants.HTTP;
} else {
String msg = "Cannot load balance for non-HTTP/S transport " + transport;
log.error(msg);
@@ -582,7 +582,7 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
}
String address = synCtx.getTo().getAddress();
- if (address.startsWith(Constants.HTTP + "://") || address.startsWith(Constants.HTTPS + "://")) {
+ if (address.startsWith(LoadBalancerConstants.HTTP + "://") || address.startsWith(LoadBalancerConstants.HTTPS + "://")) {
// Remove protocol, hostname and port found in address
try {
URL addressUrl = new URL(address);
@@ -611,7 +611,7 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
}
String memberHostName = currentMember.getHostName();
- int memberPort = (transport.startsWith(Constants.HTTPS)) ? currentMember.getHttpsPort() : currentMember.getHttpPort();
+ int memberPort = (transport.startsWith(LoadBalancerConstants.HTTPS)) ? currentMember.getHttpsPort() : currentMember.getHttpPort();
return new EndpointReference(new URL(transport, memberHostName, memberPort, address).toString());
} catch (MalformedURLException e) {
@@ -735,7 +735,7 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
private void incrementInFlightRequestCount(MessageContext messageContext) {
try {
- String clusterId = (String) messageContext.getProperty(Constants.CLUSTER_ID);
+ String clusterId = (String) messageContext.getProperty(LoadBalancerConstants.CLUSTER_ID);
if (StringUtils.isBlank(clusterId)) {
throw new RuntimeException("Cluster id not found in message context");
}
@@ -750,7 +750,7 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
private void decrementInFlightRequestCount(MessageContext messageContext) {
try {
- String clusterId = (String) messageContext.getProperty(Constants.CLUSTER_ID);
+ String clusterId = (String) messageContext.getProperty(LoadBalancerConstants.CLUSTER_ID);
if (StringUtils.isBlank(clusterId)) {
throw new RuntimeException("Cluster id not found in message context");
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalancerServiceComponent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalancerServiceComponent.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalancerServiceComponent.java
index b1ed332..7e55163 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalancerServiceComponent.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalancerServiceComponent.java
@@ -39,6 +39,7 @@ import org.apache.stratos.load.balancer.conf.configurator.SynapseConfigurator;
import org.apache.stratos.load.balancer.conf.configurator.TopologyFilterConfigurator;
import org.apache.stratos.load.balancer.context.LoadBalancerContext;
import org.apache.stratos.load.balancer.statistics.LoadBalancerStatisticsCollector;
+import org.apache.stratos.load.balancer.util.LoadBalancerConstants;
import org.apache.stratos.messaging.message.filter.topology.TopologyClusterFilter;
import org.apache.stratos.messaging.message.filter.topology.TopologyMemberFilter;
import org.apache.stratos.messaging.message.filter.topology.TopologyServiceFilter;
@@ -86,16 +87,16 @@ import java.util.concurrent.ExecutorService;
*/
@SuppressWarnings({"UnusedDeclaration", "JavaDoc"})
public class LoadBalancerServiceComponent {
+
private static final Log log = LogFactory.getLog(LoadBalancerServiceComponent.class);
private boolean activated = false;
+ private ExecutorService executorService;
private LoadBalancerTopologyEventReceiver topologyReceiver;
private LoadBalancerTenantEventReceiver tenantReceiver;
private LoadBalancerApplicationSignUpEventReceiver applicationSignUpEventReceiver;
private LoadBalancerDomainMappingEventReceiver domainMappingEventReceiver;
private LoadBalancerStatisticsNotifier statisticsNotifier;
- private static final String LOAD_BALANCER_IDENTIFIER = "Load-Balancer";
- private static final int THREAD_POOL_SIZE = 20;
protected void activate(ComponentContext ctxt) {
try {
@@ -123,7 +124,10 @@ public class LoadBalancerServiceComponent {
// Configure topology filters
TopologyFilterConfigurator.configure(configuration);
- ExecutorService executorService = StratosThreadPool.getExecutorService(LOAD_BALANCER_IDENTIFIER,THREAD_POOL_SIZE);
+ int threadPoolSize = Integer.getInteger(LoadBalancerConstants.LOAD_BALANCER_THREAD_POOL_SIZE_KEY,
+ LoadBalancerConstants.LOAD_BALANCER_DEFAULT_THREAD_POOL_SIZE);
+ executorService = StratosThreadPool.getExecutorService(LoadBalancerConstants.LOAD_BALANCER_THREAD_POOL_ID,
+ threadPoolSize);
startApplicationSignUpEventReceiver(executorService);
startDomainMappingEventReceiver(executorService);
@@ -277,16 +281,38 @@ public class LoadBalancerServiceComponent {
.getSynapseEnvironment());
}
} catch (Exception e) {
- log.warn("Couldn't remove the endpoint deployer");
+ log.warn("An error occurred while removing endpoint deployer", e);
}
+
// Terminate tenant receiver
- tenantReceiver.terminate();
+ try {
+ tenantReceiver.terminate();
+ } catch (Exception e) {
+ log.warn("An error occurred while terminating tenant event receiver", e);
+ }
+
// Terminate topology receiver
- topologyReceiver.terminate();
- // Terminate statistics notifier
- statisticsNotifier.terminate();
+ try {
+ topologyReceiver.terminate();
+ } catch (Exception e) {
+ log.warn("An error occurred while terminating topology event receiver", e);
+ }
+ // Terminate statistics notifier
+ try {
+ statisticsNotifier.terminate();
+ } catch (Exception e) {
+ log.warn("An error occurred while terminating health statistics notifier", e);
+ }
+ // Shutdown executor service
+ if(executorService != null) {
+ try {
+ executorService.shutdownNow();
+ } catch (Exception e) {
+ log.warn("An error occurred while shutting down load balancer executor service", e);
+ }
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/LocationReWriter.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/LocationReWriter.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/LocationReWriter.java
index 02253ef..88ec7bf 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/LocationReWriter.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/LocationReWriter.java
@@ -21,7 +21,7 @@ package org.apache.stratos.load.balancer.mediators;
import org.apache.commons.lang3.StringUtils;
import org.apache.stratos.load.balancer.conf.LoadBalancerConfiguration;
import org.apache.stratos.load.balancer.context.LoadBalancerContext;
-import org.apache.stratos.load.balancer.util.Constants;
+import org.apache.stratos.load.balancer.util.LoadBalancerConstants;
import org.apache.synapse.MessageContext;
import org.apache.synapse.core.axis2.Axis2MessageContext;
import org.apache.synapse.mediators.AbstractMediator;
@@ -84,9 +84,9 @@ public class LocationReWriter extends AbstractMediator {
int targetPort = -1;
if (HTTP.equals(inLocationUrl.getProtocol())) {
- targetPort = Integer.valueOf((String) messageContext.getProperty(Constants.LB_HTTP_PORT));
+ targetPort = Integer.valueOf((String) messageContext.getProperty(LoadBalancerConstants.LB_HTTP_PORT));
} else if (HTTPS.equals(inLocationUrl.getProtocol())) {
- targetPort = Integer.valueOf((String) messageContext.getProperty(Constants.LB_HTTPS_PORT));
+ targetPort = Integer.valueOf((String) messageContext.getProperty(LoadBalancerConstants.LB_HTTPS_PORT));
} else {
if (log.isWarnEnabled()) {
log.warn(String.format("An unknown protocol found: %s", inLocationUrl.getProtocol()));
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java
index 3913248..91cea19 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java
@@ -21,7 +21,7 @@ package org.apache.stratos.load.balancer.mediators;
import org.apache.commons.lang3.StringUtils;
import org.apache.stratos.load.balancer.statistics.InFlightRequestDecrementCallable;
import org.apache.stratos.load.balancer.statistics.LoadBalancerStatisticsExecutor;
-import org.apache.stratos.load.balancer.util.Constants;
+import org.apache.stratos.load.balancer.util.LoadBalancerConstants;
import org.apache.synapse.ManagedLifecycle;
import org.apache.synapse.MessageContext;
import org.apache.synapse.core.SynapseEnvironment;
@@ -46,7 +46,7 @@ public class ResponseInterceptor extends AbstractMediator implements ManagedLife
if (log.isDebugEnabled()) {
log.debug("Response interceptor mediation started");
}
- String clusterId = (String) messageContext.getProperty(Constants.CLUSTER_ID);
+ String clusterId = (String) messageContext.getProperty(LoadBalancerConstants.CLUSTER_ID);
if (StringUtils.isNotBlank(clusterId)) {
FutureTask<Object> task = new FutureTask<Object>(new InFlightRequestDecrementCallable(clusterId));
LoadBalancerStatisticsExecutor.getInstance().getService().submit(task);
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTopologyEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTopologyEventReceiver.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTopologyEventReceiver.java
index 9640ded..0c8ee28 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTopologyEventReceiver.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTopologyEventReceiver.java
@@ -35,26 +35,20 @@ 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 java.util.concurrent.ExecutorService;
-
/**
* Load balancer topology receiver updates load balancer context according to
* incoming topology events.
*/
-public class LoadBalancerTopologyEventReceiver {
+public class LoadBalancerTopologyEventReceiver extends TopologyEventReceiver {
private static final Log log = LogFactory.getLog(LoadBalancerTopologyEventReceiver.class);
- private ExecutorService executorService;
- private TopologyEventReceiver topologyEventReceiver;
public LoadBalancerTopologyEventReceiver() {
- this.topologyEventReceiver = new TopologyEventReceiver();
addEventListeners();
}
public void execute() {
- topologyEventReceiver.setExecutorService(executorService);
- topologyEventReceiver.execute();
+ super.execute();
if (log.isInfoEnabled()) {
log.info("Load balancer topology receiver thread started");
}
@@ -62,12 +56,12 @@ public class LoadBalancerTopologyEventReceiver {
private void addEventListeners() {
// Listen to topology events that affect clusters
- topologyEventReceiver.addEventListener(new CompleteTopologyEventListener() {
+ addEventListener(new CompleteTopologyEventListener() {
private boolean initialized;
@Override
protected void onEvent(Event event) {
- if(!initialized) {
+ if (!initialized) {
try {
TopologyManager.acquireReadLock();
for (Service service : TopologyManager.getTopology().getServices()) {
@@ -105,7 +99,7 @@ public class LoadBalancerTopologyEventReceiver {
return false;
}
});
- topologyEventReceiver.addEventListener(new MemberActivatedEventListener() {
+ addEventListener(new MemberActivatedEventListener() {
@Override
protected void onEvent(Event event) {
@@ -144,7 +138,7 @@ public class LoadBalancerTopologyEventReceiver {
}
// Add member to member-ip -> hostname map
- addMemberIpsToMemberIpHostnameMap(cluster, member);
+ addMemberIpsToMemberIpHostnameMap(cluster, member);
if (LoadBalancerContext.getInstance().getClusterIdClusterMap().containsCluster(
member.getClusterId())) {
@@ -168,7 +162,7 @@ public class LoadBalancerTopologyEventReceiver {
}
}
});
- topologyEventReceiver.addEventListener(new MemberMaintenanceListener() {
+ addEventListener(new MemberMaintenanceListener() {
@Override
protected void onEvent(Event event) {
@@ -195,7 +189,7 @@ public class LoadBalancerTopologyEventReceiver {
}
}
});
- topologyEventReceiver.addEventListener(new MemberSuspendedEventListener() {
+ addEventListener(new MemberSuspendedEventListener() {
@Override
protected void onEvent(Event event) {
@@ -220,7 +214,7 @@ public class LoadBalancerTopologyEventReceiver {
}
}
});
- topologyEventReceiver.addEventListener(new MemberTerminatedEventListener() {
+ addEventListener(new MemberTerminatedEventListener() {
@Override
protected void onEvent(Event event) {
@@ -246,7 +240,7 @@ public class LoadBalancerTopologyEventReceiver {
}
}
});
- topologyEventReceiver.addEventListener(new ClusterRemovedEventListener() {
+ addEventListener(new ClusterRemovedEventListener() {
@Override
protected void onEvent(Event event) {
@@ -283,7 +277,7 @@ public class LoadBalancerTopologyEventReceiver {
}
}
});
- topologyEventReceiver.addEventListener(new ServiceRemovedEventListener() {
+ addEventListener(new ServiceRemovedEventListener() {
@Override
protected void onEvent(Event event) {
@@ -399,19 +393,4 @@ public class LoadBalancerTopologyEventReceiver {
}
}
}
-
- /**
- * Terminate load balancer topology receiver thread.
- */
- public void terminate() {
- topologyEventReceiver.terminate();
- }
-
- public ExecutorService getExecutorService() {
- return executorService;
- }
-
- public void setExecutorService(ExecutorService executorService) {
- this.executorService = executorService;
- }
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsExecutor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsExecutor.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsExecutor.java
index 3351cd6..04a0756 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsExecutor.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsExecutor.java
@@ -20,23 +20,26 @@ package org.apache.stratos.load.balancer.statistics;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.common.threading.StratosThreadPool;
+import org.apache.stratos.load.balancer.util.LoadBalancerConstants;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
/**
* An executor service to asynchronously execute statistics update calls without blocking the
* mediation flow.
*/
public class LoadBalancerStatisticsExecutor {
+
private static final Log log = LogFactory.getLog(LoadBalancerStatisticsExecutor.class);
+
private static volatile LoadBalancerStatisticsExecutor instance;
- private ExecutorService service;
+ private ExecutorService executorService;
private LoadBalancerStatisticsExecutor() {
- // TODO: Move thread pool count to loadbalancer.conf
- service = Executors.newFixedThreadPool(10);
+ executorService = StratosThreadPool.getExecutorService(LoadBalancerConstants.LOAD_BALANCER_THREAD_POOL_ID,
+ LoadBalancerConstants.LOAD_BALANCER_DEFAULT_THREAD_POOL_SIZE);
}
public static LoadBalancerStatisticsExecutor getInstance() {
@@ -51,6 +54,6 @@ public class LoadBalancerStatisticsExecutor {
}
public ExecutorService getService() {
- return service;
+ return executorService;
}
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/Constants.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/Constants.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/Constants.java
deleted file mode 100644
index 6650af9..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/Constants.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.load.balancer.util;
-
-public class Constants {
-
- public static final String CLUSTER_ID = "cluster_id";
- public static final String MEMBER_ID = "member_id";
-
- public static final String HTTP = "http";
- public static final String HTTPS = "https";
-
- public static final String AXIS2_MSG_CTX_TRANSPORT_IN_URL = "TransportInURL";
- public static final String AXIS2_MSG_CTX_SERVICE_PREFIX = "SERVICE_PREFIX";
-
- public static final String LB_TARGET_HOSTNAME = "LB_TARGET_HOSTNAME";
- public static final String LB_HTTP_PORT = "LB_HTTP_PORT";
- public static final String LB_HTTPS_PORT = "LB_HTTPS_PORT";
-
- public static final String ALGORITHM_CONTEXT_CACHE = "algorithm.context.cache";
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/LoadBalancerConstants.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/LoadBalancerConstants.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/LoadBalancerConstants.java
new file mode 100644
index 0000000..2ddc7fa
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/LoadBalancerConstants.java
@@ -0,0 +1,40 @@
+/*
+ * 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.util;
+
+public class LoadBalancerConstants {
+
+ public static final String CLUSTER_ID = "cluster_id";
+ public static final String MEMBER_ID = "member_id";
+
+ public static final String HTTP = "http";
+ public static final String HTTPS = "https";
+
+ public static final String AXIS2_MSG_CTX_TRANSPORT_IN_URL = "TransportInURL";
+ public static final String AXIS2_MSG_CTX_SERVICE_PREFIX = "SERVICE_PREFIX";
+
+ public static final String LB_TARGET_HOSTNAME = "LB_TARGET_HOSTNAME";
+ public static final String LB_HTTP_PORT = "LB_HTTP_PORT";
+ public static final String LB_HTTPS_PORT = "LB_HTTPS_PORT";
+
+ public static final String LOAD_BALANCER_THREAD_POOL_ID = "load.balancer.thread.pool";
+ public static final String LOAD_BALANCER_THREAD_POOL_SIZE_KEY = "load.balancer.thread.pool.size";
+ public static final int LOAD_BALANCER_DEFAULT_THREAD_POOL_SIZE = 20;
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/internal/StratosManagerServiceComponent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/internal/StratosManagerServiceComponent.java b/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/internal/StratosManagerServiceComponent.java
index 54874b3..9eeded2 100644
--- a/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/internal/StratosManagerServiceComponent.java
+++ b/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/internal/StratosManagerServiceComponent.java
@@ -73,7 +73,7 @@ import java.util.concurrent.ExecutorService;
public class StratosManagerServiceComponent {
private static final Log log = LogFactory.getLog(StratosManagerServiceComponent.class);
- private static final String THREAD_EXECUTOR_ID = "STRATOS_MANAGER";
+ private static final String THREAD_EXECUTOR_ID = "stratos.manager.thread.pool";
private static final String STRATOS_MANAGER_COORDINATOR_LOCK = "STRATOS_MANAGER_COORDINATOR_LOCK";
private static final int THREAD_POOL_SIZE = 20;
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/messaging/receiver/StratosManagerTopologyEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/messaging/receiver/StratosManagerTopologyEventReceiver.java b/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/messaging/receiver/StratosManagerTopologyEventReceiver.java
index fc57756..1008a77 100644
--- a/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/messaging/receiver/StratosManagerTopologyEventReceiver.java
+++ b/components/org.apache.stratos.manager/src/main/java/org/apache/stratos/manager/messaging/receiver/StratosManagerTopologyEventReceiver.java
@@ -21,25 +21,13 @@ package org.apache.stratos.manager.messaging.receiver;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-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 java.util.concurrent.ExecutorService;
public class StratosManagerTopologyEventReceiver extends TopologyEventReceiver {
private static final Log log = LogFactory.getLog(StratosManagerTopologyEventReceiver.class);
public StratosManagerTopologyEventReceiver() {
- addEventListeners();
- }
-
- private void addEventListeners() {
}
@Override
http://git-wip-us.apache.org/repos/asf/stratos/blob/a6034069/extensions/cep/stratos-cep-extension/src/main/java/org/apache/stratos/cep/extension/FaultHandlingWindowProcessor.java
----------------------------------------------------------------------
diff --git a/extensions/cep/stratos-cep-extension/src/main/java/org/apache/stratos/cep/extension/FaultHandlingWindowProcessor.java b/extensions/cep/stratos-cep-extension/src/main/java/org/apache/stratos/cep/extension/FaultHandlingWindowProcessor.java
index 77d2d56..a49482b 100644
--- a/extensions/cep/stratos-cep-extension/src/main/java/org/apache/stratos/cep/extension/FaultHandlingWindowProcessor.java
+++ b/extensions/cep/stratos-cep-extension/src/main/java/org/apache/stratos/cep/extension/FaultHandlingWindowProcessor.java
@@ -60,10 +60,14 @@ import java.util.concurrent.TimeUnit;
@SiddhiExtension(namespace = "stratos", function = "faultHandling")
public class FaultHandlingWindowProcessor extends WindowProcessor implements RunnableWindowProcessor {
- private static final int TIME_OUT = 60 * 1000;
- static final Logger log = Logger.getLogger(FaultHandlingWindowProcessor.class);
- public static final String IDENTIFIER = "AutoScaler";
- private ScheduledExecutorService faultHandleScheduler;
+ private static final Logger log = Logger.getLogger(FaultHandlingWindowProcessor.class);
+
+ private static final int TIME_OUT = 60 * 1000;
+ public static final String CEP_EXTENSION_THREAD_POOL_KEY = "cep.extension.thread.pool";
+ public static final int CEP_EXTENSION_THREAD_POOL_SIZE = 10;
+
+ private ExecutorService executorService;
+ private ScheduledExecutorService faultHandleScheduler;
private ThreadBarrier threadBarrier;
private long timeToKeep;
private ISchedulerSiddhiQueue<StreamEvent> window;
@@ -278,7 +282,8 @@ public class FaultHandlingWindowProcessor extends WindowProcessor implements Run
}
MemberFaultEventMap.put("org.apache.stratos.messaging.event.health.stat.MemberFaultEvent", memberFaultEventMessageMap);
- ExecutorService executorService= StratosThreadPool.getExecutorService("AutoScaler",10);
+ executorService = StratosThreadPool.getExecutorService(CEP_EXTENSION_THREAD_POOL_KEY,
+ CEP_EXTENSION_THREAD_POOL_SIZE);
cepTopologyEventReceiver.setExecutorService(executorService);
executorService.execute(cepTopologyEventReceiver);
@@ -316,6 +321,15 @@ public class FaultHandlingWindowProcessor extends WindowProcessor implements Run
// terminate topology listener thread
cepTopologyEventReceiver.terminate();
window = null;
+
+ // Shutdown executor service
+ if(executorService != null) {
+ try {
+ executorService.shutdownNow();
+ } catch (Exception e) {
+ log.warn("An error occurred while shutting down cep extension executor service", e);
+ }
+ }
}
public ConcurrentHashMap<String, Long> getMemberTimeStampMap() {