You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stratos.apache.org by la...@apache.org on 2013/06/29 08:20:45 UTC
[46/52] [partial] initial commit to apache incubator. Here just
commit WSO2 Stratos code base as it is. After that will start re-factor code
to apache
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/main/resources/synapse-service-messages-in-flight.xml
----------------------------------------------------------------------
diff --git a/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/main/resources/synapse-service-messages-in-flight.xml b/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/main/resources/synapse-service-messages-in-flight.xml
new file mode 100644
index 0000000..7332d38
--- /dev/null
+++ b/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/main/resources/synapse-service-messages-in-flight.xml
@@ -0,0 +1,108 @@
+<?xml version="1.0" encoding="ISO-8859-1"?>
+<!--
+ ~ Copyright (c) 2005-2010, WSO2 Inc. (http://wso2.com) All Rights Reserved.
+ ~
+ ~ WSO2 Inc. 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.
+ -->
+
+<!-- The default synapse configuration shipped with the WSO2 Load Balancer
+ This handles the load balancing and the autoscaling of the stratos services
+ -->
+
+<definitions xmlns="http://ws.apache.org/ns/synapse">
+
+ <!-- You can add any flat sequences, endpoints, etc.. to this synapse-messages-in-flight.xml file if you do
+ *not* want to have the defaults given below, specific to WSO2 LB and auto-scaler
+ -->
+
+
+ <!-- Given below is the auto-scale mediator specific task. Uncomment it, if you want to
+ auto-scale your applications.
+ -->
+ <task class="org.wso2.carbon.mediator.autoscale.lbautoscale.ServiceRequestsInFlightAutoscaler"
+ name="LoadAnalyzer">
+
+ <!--
+ The private key for ec2
+ -->
+ <property name="ec2PrivateKey" value="/mnt/payload/pk.pem"/>
+
+ <!--
+ The certificate for ec2
+ -->
+ <property name="ec2Cert" value="/mnt/payload/cert.pem"/>
+ <!--
+ The key pair
+ -->
+ <property name="sshKey" value="stratos-1.0.0-keypair"/>
+
+ <property name="loadBalancer">
+ <loadBalancer>
+ <property name="securityGroup" value="stratos-appserver-lb"/>
+ <property name="instanceType" value="m1.large"/>
+ <property name="instances" value="1"/>
+ <property name="elasticIP" value="${ELASTIC_IP}"/>
+ <property name="availabilityZone" value="us-east-1c"/>
+ <property name="payload" value="/mnt/payload.zip"/>
+ </loadBalancer>
+ </property>
+
+ <property name="services">
+ <services>
+ <defaults>
+ <property name="payload" value="resources/cluster_node.zip"/>
+ <property name="availabilityZone" value="us-east-1c"/>
+ <property name="securityGroup" value="as-2011-02-23"/>
+ <property name="instanceType" value="m1.large"/>
+ <property name="minAppInstances" value="1"/>
+ <property name="maxAppInstances" value="5"/>
+ <property name="queueLengthPerNode" value="400"/>
+ <property name="roundsToAverage" value="10"/>
+ <property name="instancesPerScaleUp" value="1"/>
+ <property name="messageExpiryTime" value="60000"/>
+ </defaults>
+ <service domain="wso2.as.domain">
+ <property name="payload" value="resources/cluster_node.zip"/>
+ <property name="availabilityZone" value="us-east-1c"/>
+ </service>
+ <service domain="wso2.ds.domain">
+ <property name="payload" value="resources/cluster_node.zip"/>
+ <property name="minAppInstances" value="1"/>
+ <property name="maxAppInstances" value="5"/>
+ <property name="queueLengthPerNode" value="400"/>
+ <property name="roundsToAverage" value="10"/>
+ <property name="instancesPerScaleUp" value="1"/>
+ <property name="availabilityZone" value="us-east-1c"/>
+ <property name="securityGroup" value="ds-2011-02-23"/>
+ </service>
+ <service domain="wso2.bps.domain">
+ <property name="payload" value="resources/cluster_node.zip"/>
+ <property name="minAppInstances" value="1"/>
+ <property name="maxAppInstances" value="5"/>
+ <property name="queueLengthPerNode" value="400"/>
+ <property name="roundsToAverage" value="10"/>
+ <property name="instancesPerScaleUp" value="1"/>
+ <property name="availabilityZone" value="us-east-1c"/>
+ <property name="securityGroup" value="bps-2011-02-23"/>
+ </service>
+ </services>
+ </property>
+
+ <!--
+ The interval to trigger the task in seconds
+ -->
+ <trigger interval="5"/>
+ </task>
+</definitions>
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/java/org/wso2/carbon/mediator/autoscale/lbautoscale/AppDomainContextsTest.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/java/org/wso2/carbon/mediator/autoscale/lbautoscale/AppDomainContextsTest.java b/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/java/org/wso2/carbon/mediator/autoscale/lbautoscale/AppDomainContextsTest.java
new file mode 100644
index 0000000..1d017a3
--- /dev/null
+++ b/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/java/org/wso2/carbon/mediator/autoscale/lbautoscale/AppDomainContextsTest.java
@@ -0,0 +1,105 @@
+/*
+* Copyright (c) 2005-2011, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+*
+* WSO2 Inc. 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.wso2.carbon.mediator.autoscale.lbautoscale;
+
+import java.io.File;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.axis2.clustering.ClusteringAgent;
+import org.apache.axis2.clustering.tribes.TribesClusteringAgent;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.context.ConfigurationContextFactory;
+import org.wso2.carbon.lb.common.conf.LoadBalancerConfiguration;
+import org.wso2.carbon.lb.common.conf.LoadBalancerConfiguration.ServiceConfiguration;
+import org.wso2.carbon.lb.common.group.mgt.SubDomainAwareGroupManagementAgent;
+import org.wso2.carbon.mediator.autoscale.lbautoscale.context.AppDomainContext;
+import org.wso2.carbon.mediator.autoscale.lbautoscale.util.AutoscaleUtil;
+
+import junit.framework.Assert;
+import junit.framework.TestCase;
+
+public class AppDomainContextsTest extends TestCase {
+
+ private static Map<String, Map<String, ?>> map;
+ private LoadBalancerConfiguration lbConfig;
+ ConfigurationContext configCtx;
+ ClusteringAgent clusteringAgent;
+
+ protected void setUp() throws Exception {
+ super.setUp();
+ configCtx = ConfigurationContextFactory.createEmptyConfigurationContext();
+ clusteringAgent = new TribesClusteringAgent();
+ clusteringAgent.addGroupManagementAgent(new SubDomainAwareGroupManagementAgent(
+ "worker"),
+ "wso2.as1.domain", "worker", -1);
+ clusteringAgent.addGroupManagementAgent(new SubDomainAwareGroupManagementAgent(
+ "mgt"),
+ "wso2.as1.domain", "mgt", -1);
+ clusteringAgent.addGroupManagementAgent(new SubDomainAwareGroupManagementAgent(
+ "mgt"),
+ "wso2.as2.domain", "mgt", -1);
+ configCtx.getAxisConfiguration().setClusteringAgent(clusteringAgent);
+
+ File f = new File("src/test/resources/loadbalancer.conf");
+ System.setProperty("loadbalancer.conf", f.getAbsolutePath());
+ lbConfig = LoadBalancerConfiguration.getInstance();
+
+ map = AutoscaleUtil.getAppDomainContexts(configCtx, lbConfig);
+
+ }
+
+
+ public void testRemoval(){
+ // removing a cluster domain with only 1 sub domain
+ lbConfig.removeServiceConfiguration("wso2.as2.domain", "mgt");
+ map = AutoscaleUtil.getAppDomainContexts(configCtx, lbConfig);
+
+ Assert.assertEquals(true, !map.containsKey("wso2.as2.domain"));
+
+ // removing a cluster domain with more than 1 sub domain
+ lbConfig.removeServiceConfiguration("wso2.as1.domain", "mgt");
+ map = AutoscaleUtil.getAppDomainContexts(configCtx, lbConfig);
+
+ Assert.assertEquals(true, map.containsKey("wso2.as1.domain"));
+ Assert.assertEquals(true, map.get("wso2.as1.domain").get("mgt") == null);
+ Assert.assertEquals(true, map.get("wso2.as1.domain").get("worker") != null);
+ }
+
+ public void testAddition(){
+ ServiceConfiguration config1 = lbConfig.new ServiceConfiguration();
+ config1.setDomain("wso2.as3.domain");
+ config1.setSub_domain("mgt");
+ lbConfig.addServiceConfiguration(config1);
+ clusteringAgent.addGroupManagementAgent(new SubDomainAwareGroupManagementAgent(
+ "mgt"),
+ "wso2.as3.domain", "mgt", -1);
+ map = AutoscaleUtil.getAppDomainContexts(configCtx, lbConfig);
+
+ Assert.assertEquals(true, map.containsKey("wso2.as3.domain"));
+ Assert.assertEquals(true, map.get("wso2.as3.domain").get("mgt") != null);
+ }
+
+ @Deprecated // use only for writing test cases
+ void printKeys(Map<?,?> map){
+ for (Iterator iterator = map.keySet().iterator(); iterator.hasNext();) {
+ Object type = iterator.next();
+ System.out.println(type);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/java/org/wso2/carbon/mediator/autoscale/lbautoscale/ServiceRequestsInFlightAutoscalerTest.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/java/org/wso2/carbon/mediator/autoscale/lbautoscale/ServiceRequestsInFlightAutoscalerTest.java b/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/java/org/wso2/carbon/mediator/autoscale/lbautoscale/ServiceRequestsInFlightAutoscalerTest.java
new file mode 100644
index 0000000..1f1a084
--- /dev/null
+++ b/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/java/org/wso2/carbon/mediator/autoscale/lbautoscale/ServiceRequestsInFlightAutoscalerTest.java
@@ -0,0 +1,29 @@
+/*
+* Copyright (c) 2005-2011, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+*
+* WSO2 Inc. 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.wso2.carbon.mediator.autoscale.lbautoscale;
+
+/**
+ * Test for ServiceRequestsInFlightAutoscaler
+ */
+//public class ServiceRequestsInFlightAutoscalerTest extends TestCase {
+//
+// public void testInit() {
+//
+// //new ServiceRequestsInFlightAutoscaler().init(new Axis2SynapseEnvironment());
+// }
+//}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/resources/loadbalancer.conf
----------------------------------------------------------------------
diff --git a/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/resources/loadbalancer.conf b/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/resources/loadbalancer.conf
new file mode 100644
index 0000000..7b2a80a
--- /dev/null
+++ b/components/load-balancer/autoscaler/org.wso2.carbon.mediator.autoscale/4.1.3/src/test/resources/loadbalancer.conf
@@ -0,0 +1,50 @@
+loadbalancer {
+ instances 1;
+ enable_autoscaler true;
+ # interval between two task executions in milliseconds
+ autoscaler_task_interval 5000;
+ # after an instance booted up, task will wait till this much of time and let the server started up
+ server_startup_delay 15000; #default will be 60000ms
+}
+
+services {
+ defaults {
+ min_app_instances 1;
+ max_app_instances 5;
+ max_requests_per_second 400;
+ alarming_upper_rate 0.65;
+ alarming_lower_rate 0.2;
+ scale_down_factor 0.25;
+ rounds_to_average 10;
+ instances_per_scale_up 1;
+ message_expiry_time 60000;
+ }
+
+ appserver {
+ hosts appserver.cloud-test.wso2.com, as.cloud-test.wso2.com;
+ sub_domain worker1;
+ domains {
+ wso2.as1.domain {
+ hosts as1.cloud-test.wso2.com;
+ min_app_instances 0;
+ sub_domain worker;
+ tenant_range 1-5;
+ }
+ wso2.as1.domain {
+ hosts mgt.as1.cloud-test.wso2.com;
+ min_app_instances 0;
+ sub_domain mgt;
+ tenant_range 1-5;
+ }
+ wso2.as2.domain {
+ hosts mgt.as2.cloud-test.wso2.com;
+ min_app_instances 0;
+ sub_domain mgt;
+ tenant_range 1-5;
+ }
+
+ }
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/pom.xml
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/pom.xml b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/pom.xml
new file mode 100644
index 0000000..49f4624
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/pom.xml
@@ -0,0 +1,107 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+ ~ Copyright (c) 2009-2010, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ ~
+ ~ Licensed 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <parent>
+ <groupId>org.wso2.carbon</groupId>
+ <artifactId>loadbalancer-components</artifactId>
+ <version>4.1.0</version>
+ <relativePath>../../../pom.xml</relativePath>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>org.wso2.carbon</groupId>
+ <artifactId>org.wso2.carbon.lb.endpoint</artifactId>
+ <packaging>bundle</packaging>
+ <version>4.1.3</version>
+ <name>WSO2 Carbon - Tenant Aware LoadBalance Endpoint</name>
+ <url>http://wso2.org</url>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.wso2.carbon</groupId>
+ <artifactId>org.wso2.carbon.mediation.initializer</artifactId>
+ <version>${wso2carbon.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.wso2.carbon</groupId>
+ <artifactId>org.wso2.carbon.mediation.dependency.mgt</artifactId>
+ <version>${wso2carbon.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.synapse</groupId>
+ <artifactId>synapse-core</artifactId>
+ <version>2.1.1-wso2v4</version>
+ </dependency>
+ <dependency>
+ <groupId>org.wso2.carbon</groupId>
+ <artifactId>org.wso2.carbon.logging</artifactId>
+ <version>${wso2carbon.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.wso2.carbon</groupId>
+ <artifactId>org.wso2.carbon.lb.common</artifactId>
+ <version>4.1.3</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.axis2</groupId>
+ <artifactId>axis2-kernel</artifactId>
+ <version>1.6.1-wso2v9</version>
+ </dependency>
+ <dependency>
+ <groupId>org.wso2.carbon</groupId>
+ <artifactId>org.wso2.carbon.cartridge.messages</artifactId>
+ <version>2.1.1</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-scr-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+
+ <extensions>true</extensions>
+ <configuration>
+ <instructions>
+ <Bundle-SymbolicName>${project.artifactId}</Bundle-SymbolicName>
+ <Bundle-Name>${project.artifactId}</Bundle-Name>
+ <Export-Package>
+ org.wso2.carbon.lb.endpoint.endpoint,
+ org.wso2.carbon.lb.endpoint,
+ org.wso2.carbon.lb.endpoint.util,
+ </Export-Package>
+ <Import-Package>
+ !org.wso2.carbon.lb.endpoint.endpoint,
+ !org.wso2.carbon.lb.endpoint,
+ org.wso2.carbon.lb.common.*; version=${wso2carbon.version},
+ org.wso2.carbon.registry.core.service; version=1.0.1,
+ *;resolution:=optional
+ </Import-Package>
+ <DynamicImport-Package>*</DynamicImport-Package>
+ </instructions>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/EndpointConstants.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/EndpointConstants.java b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/EndpointConstants.java
new file mode 100644
index 0000000..cf31ef0
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/EndpointConstants.java
@@ -0,0 +1,28 @@
+/**
+ * Copyright (c) 2009, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * Licensed 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.wso2.carbon.lb.endpoint;
+
+/**
+ * The <code>EndpointConstants</code> class defined the endpoint constants.
+ */
+public class EndpointConstants {
+
+ public static final String ADDRESS_ENDPOINT = "Address Endpoint";
+ public static final String WSDL_ENDPOINT = "WSDL Endpoint";
+ public static final String FAILOVER_ENDPOINT = "Failover Group";
+ public static final String LOADBALANCE_ENDPOINT = "Loadbalance Endpoint";
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/EndpointDeployer.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/EndpointDeployer.java b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/EndpointDeployer.java
new file mode 100644
index 0000000..d9366b8
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/EndpointDeployer.java
@@ -0,0 +1,68 @@
+/**
+ * Copyright (c) 2009, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * Licensed 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.wso2.carbon.lb.endpoint;
+
+import org.apache.axiom.om.OMElement;
+import org.apache.axis2.context.ConfigurationContext;
+import org.wso2.carbon.mediation.initializer.ServiceBusConstants;
+import org.wso2.carbon.mediation.initializer.ServiceBusUtils;
+import org.wso2.carbon.mediation.initializer.persistence.MediationPersistenceManager;
+
+import java.util.Properties;
+
+/**
+ * Responsible for deploying Synapse artifacts.
+ */
+public class EndpointDeployer extends org.apache.synapse.deployers.EndpointDeployer {
+
+ MediationPersistenceManager mpm;
+
+ @Override
+ public void init(ConfigurationContext configCtx) {
+ super.init(configCtx);
+ this.mpm = ServiceBusUtils.getMediationPersistenceManager(configCtx.getAxisConfiguration());
+ }
+
+ @Override
+ public String deploySynapseArtifact(OMElement artifactConfig, String fileName,
+ Properties properties) {
+ String epName = super.deploySynapseArtifact(artifactConfig, fileName, properties);
+ mpm.saveItemToRegistry(epName, ServiceBusConstants.ITEM_TYPE_ENDPOINT);
+ return epName;
+ }
+
+ @Override
+ public String updateSynapseArtifact(OMElement artifactConfig, String fileName,
+ String existingArtifactName, Properties properties) {
+ String epName = super.updateSynapseArtifact(
+ artifactConfig, fileName, existingArtifactName, properties);
+ mpm.saveItemToRegistry(epName, ServiceBusConstants.ITEM_TYPE_ENDPOINT);
+ return epName;
+ }
+
+ @Override
+ public void undeploySynapseArtifact(String artifactName) {
+ super.undeploySynapseArtifact(artifactName);
+ mpm.deleteItemFromRegistry(artifactName, ServiceBusConstants.ITEM_TYPE_ENDPOINT);
+ }
+
+ @Override
+ public void restoreSynapseArtifact(String artifactName) {
+ super.restoreSynapseArtifact(artifactName);
+ mpm.saveItemToRegistry(artifactName, ServiceBusConstants.ITEM_TYPE_ENDPOINT);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/TenantAwareLoadBalanceEndpointException.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/TenantAwareLoadBalanceEndpointException.java b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/TenantAwareLoadBalanceEndpointException.java
new file mode 100644
index 0000000..d49758a
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/TenantAwareLoadBalanceEndpointException.java
@@ -0,0 +1,36 @@
+/**
+ * Copyright (c) 2009, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * Licensed 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.wso2.carbon.lb.endpoint;
+
+/**
+ * Exception to be thrown from this component.
+ */
+public class TenantAwareLoadBalanceEndpointException extends RuntimeException {
+
+ private static final long serialVersionUID = -663839410798538370L;
+
+ public TenantAwareLoadBalanceEndpointException(String msg) {
+ super(msg);
+ }
+
+ public TenantAwareLoadBalanceEndpointException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+
+ public TenantAwareLoadBalanceEndpointException(Throwable cause) {
+ super(cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/TenantLoadBalanceMembershipHandler.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/TenantLoadBalanceMembershipHandler.java b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/TenantLoadBalanceMembershipHandler.java
new file mode 100644
index 0000000..dec45b4
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/TenantLoadBalanceMembershipHandler.java
@@ -0,0 +1,190 @@
+package org.wso2.carbon.lb.endpoint;
+
+
+import org.apache.axis2.clustering.ClusteringAgent;
+import org.apache.axis2.clustering.Member;
+import org.apache.axis2.clustering.management.GroupManagementAgent;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.synapse.SynapseException;
+import org.apache.synapse.core.LoadBalanceMembershipHandler;
+import org.apache.synapse.endpoints.algorithms.AlgorithmContext;
+import org.apache.synapse.endpoints.algorithms.LoadbalanceAlgorithm;
+import org.wso2.carbon.lb.common.conf.util.HostContext;
+import org.wso2.carbon.lb.endpoint.util.ConfigHolder;
+import org.wso2.carbon.user.api.UserStoreException;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Bridge between Axis2 membership notification and Synapse load balancing
+ */
+public class TenantLoadBalanceMembershipHandler implements LoadBalanceMembershipHandler {
+ private static final Log log = LogFactory.getLog(TenantLoadBalanceMembershipHandler.class);
+
+ private ConfigurationContext configCtx;
+
+ private LoadbalanceAlgorithm lbAlgo;
+
+ /**
+ * Key - Host, Value - HostContext
+ */
+ private static Map<String, HostContext> hostContextsMap =
+ new HashMap<String, HostContext>();
+
+ private ClusteringAgent clusteringAgent;
+
+ private boolean isClusteringEnabled;
+ private String endpointName;
+
+ public TenantLoadBalanceMembershipHandler(Map<String, HostContext> hostContexts,
+ LoadbalanceAlgorithm algorithm,
+ ConfigurationContext configCtx,
+ boolean isClusteringEnabled,
+ String endpointName) {
+
+ lbAlgo = algorithm;
+ this.isClusteringEnabled = isClusteringEnabled;
+ this.endpointName = endpointName;
+ this.configCtx = configCtx;
+
+ for (HostContext host : hostContexts.values()) {
+
+ addHostContext(host);
+
+ }
+ }
+
+ /**
+ * This will be used to add new {@link HostContext}s.
+ * @param host {@link HostContext}
+ */
+ public void addHostContext(HostContext host){
+
+ String hostName = host.getHostName();
+
+ AlgorithmContext algorithmContext =
+ new AlgorithmContext(isClusteringEnabled,
+ configCtx, endpointName + "." +
+ hostName);
+
+ host.setAlgorithm(lbAlgo.clone());
+ host.setAlgorithmContext(algorithmContext);
+
+ hostContextsMap.put(hostName, host);
+
+ }
+
+ /**
+ * This will be used to remove an existing {@link HostContext}s.
+ * @param host {@link HostContext}
+ */
+ public void removeHostContext(String host){
+
+ hostContextsMap.remove(host);
+
+ }
+
+ public void init(Properties props, LoadbalanceAlgorithm algorithm) {
+ // Nothing to do
+ }
+
+ public void setConfigurationContext(ConfigurationContext configCtx) {
+ this.configCtx = configCtx;
+
+ // The following code does the bridging between Axis2 and Synapse load balancing
+ clusteringAgent = configCtx.getAxisConfiguration().getClusteringAgent();
+ if (clusteringAgent == null) {
+ String msg = "In order to enable load balancing across an Axis2 cluster, " +
+ "the cluster entry should be enabled in the axis2.xml file";
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ }
+
+ public ConfigurationContext getConfigurationContext() {
+ return configCtx;
+ }
+
+ /**
+ * Getting the next member to which the request has to be sent in a round-robin fashion
+ *
+ * @param context The AlgorithmContext
+ * @return The current member
+ * @deprecated Use {@link #getNextApplicationMember(String, int)}
+ */
+ public Member getNextApplicationMember(AlgorithmContext context) {
+ throw new UnsupportedOperationException("This operation is invalid. " +
+ "Call getNextApplicationMember(String host)");
+ }
+
+ public boolean isAValidHostName(String host){
+ if(getHostContext(host) != null){
+ return true;
+ }
+ return false;
+ }
+
+ public Member getNextApplicationMember(String host, int tenantId) {
+ HostContext hostContext = getHostContext(host);
+
+ if(hostContext == null){
+ String msg = "Invalid host name : " + host;
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+
+ // here we have to pass tenant id to get domain from hostContext
+ String domain = hostContext.getDomainFromTenantId(tenantId);
+ String subDomain = hostContext.getSubDomainFromTenantId(tenantId);
+
+ LoadbalanceAlgorithm algorithm = hostContext.getAlgorithm();
+ GroupManagementAgent groupMgtAgent = clusteringAgent.getGroupManagementAgent(domain, subDomain);
+
+ if (groupMgtAgent == null) {
+ String tenantDomain;
+ try {
+ tenantDomain = ConfigHolder.getInstance().getRealmService().getTenantManager().getDomain(tenantId);
+ } catch (UserStoreException ignore) {
+ tenantDomain = ""+tenantId;
+ }
+
+ String msg =
+ "No Group Management Agent found for the domain: " + domain + ", subDomain: "
+ + subDomain + ", host: " + host+ " and for tenant: "
+ + tenantDomain;
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ algorithm.setApplicationMembers(groupMgtAgent.getMembers());
+ AlgorithmContext context = hostContext.getAlgorithmContext();
+ return algorithm.getNextApplicationMember(context);
+ }
+
+ public HostContext getHostContext(String host) {
+ HostContext hostContext = hostContextsMap.get(host);
+ if (hostContext == null) {
+ int indexOfDot;
+ if ((indexOfDot = host.indexOf(".")) != -1) {
+ hostContext = getHostContext(host.substring(indexOfDot + 1));
+ }
+ }
+ return hostContext;
+ }
+
+ public LoadbalanceAlgorithm getLoadbalanceAlgorithm() {
+ return lbAlgo;
+ }
+
+ public Properties getProperties() {
+ return null;
+ }
+
+ public ClusteringAgent getClusteringAgent() {
+ return clusteringAgent;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/builder/TopologySyncher.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/builder/TopologySyncher.java b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/builder/TopologySyncher.java
new file mode 100644
index 0000000..31e9f67
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/builder/TopologySyncher.java
@@ -0,0 +1,142 @@
+package org.wso2.carbon.lb.endpoint.builder;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.synapse.SynapseException;
+import org.wso2.carbon.lb.common.conf.LoadBalancerConfiguration;
+import org.wso2.carbon.lb.common.conf.LoadBalancerConfiguration.ServiceConfiguration;
+import org.wso2.carbon.lb.common.conf.structure.Node;
+import org.wso2.carbon.lb.common.conf.structure.NodeBuilder;
+import org.wso2.carbon.lb.common.conf.util.HostContext;
+import org.wso2.carbon.lb.common.conf.util.TenantDomainContext;
+import org.wso2.carbon.lb.endpoint.TenantLoadBalanceMembershipHandler;
+import org.wso2.carbon.lb.endpoint.group.mgt.GroupMgtAgentBuilder;
+import org.wso2.carbon.lb.endpoint.util.ConfigHolder;
+
+/**
+ * A Thread, which is responsible for making a sense out of a message received for
+ * ELB via topology synchronization.
+ */
+public class TopologySyncher implements Runnable {
+
+ private static final Log log = LogFactory.getLog(TopologySyncher.class);
+
+ /*
+ * This is a reference to sharedTopologyQueue ConfigHolder.
+ */
+ private BlockingQueue<String> sharedQueue;
+
+ public TopologySyncher(BlockingQueue<String> queue) {
+
+ sharedQueue = queue;
+
+ }
+
+ @Override
+ public void run() {
+ // grab the lb configuration instance
+ LoadBalancerConfiguration lbconfig = LoadBalancerConfiguration.getInstance();
+
+ // FIXME Currently there has to be at least one dummy cluster defined in the loadbalancer
+ // conf
+ // in order to proper initialization of TribesClusteringAgent.
+ generateGroupMgtAgents(lbconfig);
+
+ // this thread should run for ever, untill terminated.
+ while (true) {
+ try {
+
+ // grabs a message or waits till the queue is non-empty
+ String message = sharedQueue.take();
+// ConfigHolder data = ConfigHolder.getInstance();
+
+ // this message needs attention only if it's not same as the previous message and
+ // not null of course.
+// if (data.getPreviousMsg() == null || !data.getPreviousMsg().equals(message)) {
+
+ // reset the previous message
+// data.setPreviousMsg(message);
+
+ // build the nginx format of this message, and get the Node object
+ Node topologyNode = NodeBuilder.buildNode(message);
+
+ // reset service configurations
+// lbconfig.resetData();
+ // create new service configurations
+ List<ServiceConfiguration> currentServiceConfigs = lbconfig.createServicesConfig(topologyNode);
+
+ generateGroupMgtAgents(lbconfig);
+
+ removeGroupMgtAgents(lbconfig, currentServiceConfigs);
+
+// }
+
+ } catch (InterruptedException ignore) {
+ }
+ }
+
+ }
+
+ private void removeGroupMgtAgents(LoadBalancerConfiguration lbConfig, List<ServiceConfiguration> currentServiceConfigs) {
+
+ for (Iterator iterator = lbConfig.getServiceConfigurations().values().iterator(); iterator.hasNext();) {
+ Map<String, ServiceConfiguration> valuesMap = (Map<String, ServiceConfiguration>) iterator.next();
+
+ for (Iterator iterator2 = valuesMap.values().iterator(); iterator2.hasNext();) {
+ ServiceConfiguration oldServiceConfig = (ServiceConfiguration) iterator2.next();
+
+ if(!currentServiceConfigs.contains(oldServiceConfig)){
+ // if the ServiceConfiguration is not there any more in the latest topology
+ lbConfig.removeServiceConfiguration(oldServiceConfig.getDomain(), oldServiceConfig.getSubDomain());
+ GroupMgtAgentBuilder.resetGroupMgtAgent(oldServiceConfig.getDomain(), oldServiceConfig.getSubDomain());
+ }
+ }
+ }
+ }
+
+ /**
+ * @param lbconfig
+ */
+ private void generateGroupMgtAgents(LoadBalancerConfiguration lbconfig) {
+ TenantLoadBalanceMembershipHandler handler =
+ ConfigHolder.getInstance()
+ .getTenantLoadBalanceMembershipHandler();
+
+ if (handler == null) {
+ String msg =
+ "TenantLoadBalanceMembershipHandler is null. Thus, We cannot proceed.";
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+
+ Map<String, HostContext> hostContexts = lbconfig.getHostContextMap();
+
+ // Add the Axis2 GroupManagement agents
+ if (hostContexts != null) {
+ // iterate through each host context
+ for (HostContext hostCtxt : hostContexts.values()) {
+ // each host can has multiple Tenant Contexts, iterate through them
+ for (TenantDomainContext tenantCtxt : hostCtxt
+ .getTenantDomainContexts()) {
+
+ String domain = tenantCtxt.getDomain();
+ String subDomain = tenantCtxt.getSubDomain();
+
+ // creates the group management agent
+ GroupMgtAgentBuilder.createGroupMgtAgent(domain,
+ subDomain);
+ }
+
+ // add to the handler
+ handler.addHostContext(hostCtxt);
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/cluster/manager/ClusterDomainManagerImpl.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/cluster/manager/ClusterDomainManagerImpl.java b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/cluster/manager/ClusterDomainManagerImpl.java
new file mode 100644
index 0000000..1032fc9
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/cluster/manager/ClusterDomainManagerImpl.java
@@ -0,0 +1,227 @@
+/*
+ * Copyright (c) 2005-2011, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * WSO2 Inc. 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.wso2.carbon.lb.endpoint.cluster.manager;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.axis2.clustering.ClusteringAgent;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.synapse.SynapseException;
+import org.wso2.carbon.cartridge.messages.ClusterDomainManager;
+import org.wso2.carbon.cartridge.messages.ClusterDomain;
+import org.wso2.carbon.lb.common.conf.LoadBalancerConfiguration;
+import org.wso2.carbon.lb.common.conf.LoadBalancerConfiguration.ServiceConfiguration;
+import org.wso2.carbon.lb.common.conf.util.Constants;
+import org.wso2.carbon.lb.common.conf.util.HostContext;
+import org.wso2.carbon.lb.common.conf.util.TenantDomainContext;
+import org.wso2.carbon.lb.common.group.mgt.SubDomainAwareGroupManagementAgent;
+import org.wso2.carbon.lb.endpoint.TenantLoadBalanceMembershipHandler;
+import org.wso2.carbon.lb.endpoint.util.ConfigHolder;
+
+/**
+ * Bridge between the ELB and the Stratos2 Agent.
+ */
+public class ClusterDomainManagerImpl implements ClusterDomainManager {
+
+ private static final Log log = LogFactory.getLog(ClusterDomainManagerImpl.class);
+
+ @Override
+ public void addClusterDomain(ClusterDomain cluster) {
+
+ // create group management agent, if one doesn't exist already.
+ HostContext hostCtxt = createGroupMgtAgentIfNotExists(cluster);
+
+ // we should only update if the above step is successful.
+ if (hostCtxt != null) {
+ // create / update Service Configuration
+ createOrUpdateServiceConfig(cluster, hostCtxt);
+ }
+
+ }
+
+ @Override
+ public void removeClusterDomain(String domain, String subDomain, String hostName) {
+
+ TenantLoadBalanceMembershipHandler handler =
+ ConfigHolder.getInstance()
+ .getTenantLoadBalanceMembershipHandler();
+
+ if (handler == null) {
+ String msg = "TenantLoadBalanceMembershipHandler is null. Thus, We cannot proceed.";
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+
+ handler.removeHostContext(hostName);
+
+ LoadBalancerConfiguration lbConfig = ConfigHolder.getInstance().getLbConfig();
+
+ lbConfig.removeServiceConfiguration(domain, subDomain);
+
+ }
+
+ private void createOrUpdateServiceConfig(ClusterDomain cluster, HostContext ctxt) {
+ LoadBalancerConfiguration lbConfig = ConfigHolder.getInstance().getLbConfig();
+
+ String domain = cluster.getDomain();
+ String subDomain = cluster.getSubDomain();
+
+ if (subDomain == null || subDomain.isEmpty()) {
+ // uses default sub domain
+ subDomain = Constants.DEFAULT_SUB_DOMAIN;
+ }
+
+ String hostName = cluster.getHostName();
+ String tenantRange = cluster.getTenantRange();
+ int minInstances = cluster.getMinInstances();
+ int maxInstances = cluster.getMaxInstances();
+ String service = cluster.getServiceName();
+ int maxRequestsPerSecond = cluster.getMaxRequestsPerSecond();
+ int roundsToAverage = cluster.getRoundsToAverage();
+ double alarmingUpperRate = cluster.getAlarmingUpperRate();
+ double alarmingLowerRate = cluster.getAlarmingLowerRate();
+ double scaleDownFactor = cluster.getScaleDownFactor();
+
+ ServiceConfiguration serviceConfig ;
+
+ if((serviceConfig = lbConfig.getServiceConfig(domain, subDomain)) == null){
+ serviceConfig = lbConfig.new ServiceConfiguration();
+ }
+
+ // we simply override the attributes with new values
+ serviceConfig.setDomain(domain);
+ serviceConfig.setSub_domain(subDomain);
+ serviceConfig.setTenant_range(tenantRange);
+ serviceConfig.setHosts(hostName);
+ serviceConfig.setMin_app_instances(minInstances);
+ serviceConfig.setMax_app_instances(maxInstances);
+ serviceConfig.setMax_requests_per_second(maxRequestsPerSecond);
+ serviceConfig.setRounds_to_average(roundsToAverage);
+ serviceConfig.setAlarming_upper_rate(alarmingUpperRate);
+ serviceConfig.setAlarming_lower_rate(alarmingLowerRate);
+ serviceConfig.setScale_down_factor(scaleDownFactor);
+
+ // add to host name tracker
+ lbConfig.addToHostNameTrackerMap(service, serviceConfig.getHosts());
+
+ // add to host contexts
+ lbConfig.addToHostContextMap(hostName, ctxt);
+
+ // finally add this service config
+ lbConfig.addServiceConfiguration(serviceConfig);
+ }
+
+ protected HostContext createGroupMgtAgentIfNotExists(ClusterDomain cluster) {
+
+ String domain = cluster.getDomain();
+ String subDomain = cluster.getSubDomain();
+ String hostName = cluster.getHostName();
+ String tenantRange = cluster.getTenantRange();
+
+ // sub domain can be null, but others can't
+ if (domain == null || hostName == null || tenantRange == null) {
+ String msg =
+ "Invalid value/s detected - domain: " + domain + "\n host name: " +
+ hostName + "\n tenant range: " + tenantRange;
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+
+ if (subDomain == null || subDomain.isEmpty()) {
+ // uses default sub domain
+ subDomain = Constants.DEFAULT_SUB_DOMAIN;
+ }
+
+ ClusteringAgent clusteringAgent = null;
+
+ try {
+ clusteringAgent =
+ ConfigHolder.getInstance().getAxisConfiguration()
+ .getClusteringAgent();
+
+ } catch (Exception e) {
+ String msg = "Failed to retrieve Clustering Agent.";
+ log.error(msg, e);
+ throw new SynapseException(msg, e);
+
+ }
+
+ if (clusteringAgent == null) {
+ String msg = "Clustering Agent is null.";
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+
+ /*
+ * Add Group Management Agent if one is not already present for this domain and sub
+ * domain
+ */
+
+ if (clusteringAgent.getGroupManagementAgent(domain, subDomain) == null) {
+ clusteringAgent.addGroupManagementAgent(new SubDomainAwareGroupManagementAgent(subDomain),
+ domain, subDomain,-1);
+
+ if (log.isDebugEnabled()) {
+ log.debug("Group management agent added to cluster domain: " + domain +
+ " and sub domain: " + subDomain);
+ }
+
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("Group management agent is already available for cluster domain: " +
+ domain + " and sub domain: " + subDomain);
+ }
+ }
+
+ TenantLoadBalanceMembershipHandler handler =
+ ConfigHolder.getInstance()
+ .getTenantLoadBalanceMembershipHandler();
+
+ if (handler == null) {
+ String msg = "TenantLoadBalanceMembershipHandler is null. Thus, We cannot proceed.";
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+
+ HostContext hostCtxt;
+
+ // if there's an already registered HostContext use it
+ if((hostCtxt = handler.getHostContext(hostName)) == null){
+ hostCtxt = new HostContext(hostName);
+ }
+
+ List<TenantDomainContext> ctxts;
+ ctxts = new ArrayList<TenantDomainContext>(hostCtxt.getTenantDomainContexts());
+
+ // default value is super tenant mode - which is defined by tenant id 0, in this context
+ int tenantId = 0;
+ if(!"*".equals(tenantRange)){
+ tenantId = Integer.parseInt(tenantRange);
+ }
+
+ ctxts.add(new TenantDomainContext(tenantId, domain, subDomain));
+
+ hostCtxt.addTenantDomainContexts(ctxts);
+
+ handler.addHostContext(hostCtxt);
+
+ return hostCtxt;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/endpoint/TenantAwareLoadBalanceEndpoint.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/endpoint/TenantAwareLoadBalanceEndpoint.java b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/endpoint/TenantAwareLoadBalanceEndpoint.java
new file mode 100644
index 0000000..16f392a
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/endpoint/TenantAwareLoadBalanceEndpoint.java
@@ -0,0 +1,513 @@
+package org.wso2.carbon.lb.endpoint.endpoint;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.axiom.om.OMElement;
+import org.apache.axiom.om.util.AXIOMUtil;
+import org.apache.axis2.addressing.EndpointReference;
+import org.apache.axis2.clustering.ClusteringAgent;
+import org.apache.axis2.clustering.Member;
+import org.apache.axis2.clustering.management.GroupManagementAgent;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.http.protocol.HTTP;
+import org.apache.synapse.MessageContext;
+import org.apache.synapse.SynapseConstants;
+import org.apache.synapse.SynapseException;
+import org.apache.synapse.config.xml.endpoints.utils.LoadbalanceAlgorithmFactory;
+import org.apache.synapse.core.LoadBalanceMembershipHandler;
+import org.apache.synapse.core.SynapseEnvironment;
+import org.apache.synapse.core.axis2.Axis2MessageContext;
+import org.apache.synapse.core.axis2.Axis2SynapseEnvironment;
+import org.apache.synapse.endpoints.DynamicLoadbalanceFaultHandler;
+import org.apache.synapse.endpoints.Endpoint;
+import org.apache.synapse.endpoints.algorithms.LoadbalanceAlgorithm;
+import org.apache.synapse.endpoints.dispatch.HttpSessionDispatcher;
+import org.apache.synapse.endpoints.dispatch.SALSessions;
+import org.apache.synapse.endpoints.dispatch.SessionInformation;
+import org.apache.synapse.transport.nhttp.NhttpConstants;
+import org.wso2.carbon.base.MultitenantConstants;
+import org.wso2.carbon.lb.common.conf.LoadBalancerConfiguration;
+import org.wso2.carbon.lb.common.conf.util.HostContext;
+import org.wso2.carbon.lb.common.conf.util.TenantDomainContext;
+import org.wso2.carbon.lb.common.util.DomainMapping;
+import org.wso2.carbon.lb.common.group.mgt.*;
+import org.wso2.carbon.lb.common.cache.URLMappingCache;
+import org.wso2.carbon.lb.endpoint.TenantAwareLoadBalanceEndpointException;
+import org.wso2.carbon.lb.endpoint.TenantLoadBalanceMembershipHandler;
+import org.wso2.carbon.lb.endpoint.internal.RegistryManager;
+import org.wso2.carbon.lb.endpoint.util.ConfigHolder;
+
+public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints.DynamicLoadbalanceEndpoint implements Serializable {
+
+ private static final long serialVersionUID = 1577351815951789938L;
+ private static final Log log = LogFactory.getLog(TenantAwareLoadBalanceEndpoint.class);
+ /**
+ * Axis2 based membership handler which handles members in multiple clustering domains
+ */
+ private TenantLoadBalanceMembershipHandler tlbMembershipHandler;
+
+ /**
+ * Key - host name
+ * Value - {@link HostContext}
+ */
+ private Map<String, HostContext> hostContexts = new HashMap<String, HostContext>();
+
+ private LoadBalancerConfiguration lbConfig;
+
+ /**
+ * keep the size of cache which used to keep hostNames of url mapping.
+ */
+ private URLMappingCache mappingCache;
+ private RegistryManager registryManager;
+ private int sizeOfCache;
+
+ private boolean initialized;
+
+ private String algorithm;
+ private String configuration;
+ private String failOver;
+
+ @Override
+ public void init(SynapseEnvironment synapseEnvironment) {
+ try {
+
+ lbConfig = ConfigHolder.getInstance().getLbConfig();
+ hostContexts = lbConfig.getHostContextMap();
+ sizeOfCache = lbConfig.getLoadBalancerConfig().getSizeOfCache();
+ mappingCache = URLMappingCache.getInstance(sizeOfCache);
+ setSessionTimeout(lbConfig.getLoadBalancerConfig().getSessionTimeOut());
+ setFailover(lbConfig.getLoadBalancerConfig().getFailOver());
+
+ } catch (Exception e) {
+ String msg = "Failed while reading Load Balancer configuration";
+ log.error(msg, e);
+ throw new TenantAwareLoadBalanceEndpointException(msg, e);
+ }
+
+
+ LoadbalanceAlgorithm algorithm = null;
+ try {
+ OMElement payload = AXIOMUtil.stringToOM(generatePayLoad());
+ algorithm =
+ LoadbalanceAlgorithmFactory.
+ createLoadbalanceAlgorithm(payload, null);
+
+ } catch (Exception e) {
+ String msg = "Error While creating Load balance algorithm";
+ log.error(msg, e);
+ throw new SynapseException(msg, e);
+ }
+
+ if (!initialized) {
+ super.init(synapseEnvironment);
+ ConfigurationContext cfgCtx =
+ ((Axis2SynapseEnvironment) synapseEnvironment).getAxis2ConfigurationContext();
+ ClusteringAgent clusteringAgent = cfgCtx.getAxisConfiguration().getClusteringAgent();
+ if (clusteringAgent == null) {
+ throw new SynapseException("Axis2 ClusteringAgent not defined in axis2.xml");
+ }
+
+ // Add the Axis2 GroupManagement agents
+ if (hostContexts != null) {
+ // iterate through each host context
+ for (HostContext hostCtxt : hostContexts.values()) {
+ // each host can has multiple Tenant Contexts, iterate through them
+ for (TenantDomainContext tenantCtxt : hostCtxt.getTenantDomainContexts()) {
+
+ String domain = tenantCtxt.getDomain();
+ String subDomain = tenantCtxt.getSubDomain();
+
+ if (clusteringAgent.getGroupManagementAgent(domain, subDomain) == null) {
+ String gmAgentClass = lbConfig.getLoadBalancerConfig().getGroupManagementAgentClass();
+ GroupManagementAgent groupManagementAgent;
+ if (gmAgentClass != null) {
+ try {
+ groupManagementAgent = (GroupManagementAgent) Class.forName(gmAgentClass).newInstance();
+ } catch (Exception e) {
+ String msg = "Cannot instantiate GroupManagementAgent. Class: " + gmAgentClass;
+ log.error(msg, e);
+ throw new TenantAwareLoadBalanceEndpointException(msg, e);
+ }
+ } else {
+ groupManagementAgent = new SubDomainAwareGroupManagementAgent(subDomain);
+ }
+ clusteringAgent.addGroupManagementAgent(groupManagementAgent,
+ domain, subDomain,-1);
+ if (log.isDebugEnabled()) {
+ log.debug("Group management agent added to cluster domain: " +
+ domain + " and sub domain: " + subDomain);
+ }
+ }
+ }
+ }
+
+ tlbMembershipHandler =
+ new TenantLoadBalanceMembershipHandler(hostContexts,
+ algorithm, cfgCtx,
+ isClusteringEnabled,
+ getName());
+
+ // set TenantLoadBalanceMembershipHandler for future reference
+ ConfigHolder.getInstance().setTenantLoadBalanceMembershipHandler(tlbMembershipHandler);
+ }
+
+ // Initialize the SAL Sessions if already has not been initialized.
+ SALSessions salSessions = SALSessions.getInstance();
+ if (!salSessions.isInitialized()) {
+ salSessions.initialize(isClusteringEnabled, cfgCtx);
+ }
+ setSessionAffinity(true);
+ setDispatcher(new HttpSessionDispatcher());
+ initialized = true;
+ log.info("Tenant Aware Load Balance Endpoint is initialized.");
+ }
+
+ }
+
+ public void setConfiguration(String paramEle) {
+ configuration = paramEle;
+ }
+
+ public void setAlgorithm(String paramEle) {
+ this.algorithm = paramEle;
+ }
+
+ public void setFailOver(String paramEle) {
+ this.failOver = paramEle;
+ }
+
+
+ public String getName() {
+ return "tlbEndpoint";
+ }
+
+ //TODO remove following hard coded element
+ private String generatePayLoad() {
+ return " <serviceDynamicLoadbalance failover=\"true\"\n" +
+ " algorithm=\"org.apache.synapse.endpoints.algorithms.RoundRobin\"" +
+ //" configuration=\"$system:loadbalancer.xml\"" +
+ "/>";
+ }
+
+ public LoadBalanceMembershipHandler getLbMembershipHandler() {
+ return tlbMembershipHandler;
+ }
+
+
+ public void send(MessageContext synCtx) {
+ /* setCookieHeader(synCtx); */
+ Member currentMember = null;
+ SessionInformation sessionInformation = null;
+ String actualHost = null;
+
+ //Gathering required information for domain mapping
+ org.apache.axis2.context.MessageContext axis2MessageContext =
+ ((Axis2MessageContext) synCtx).getAxis2MessageContext();
+ Map<String, String> transportHeaders = (Map<String, String>) axis2MessageContext.
+ getProperty(org.apache.axis2.context.MessageContext.TRANSPORT_HEADERS);
+ String targetHost = transportHeaders.get(HTTP.TARGET_HOST);
+
+ String port = "";
+ boolean containsPort = false;
+ if (targetHost.contains(":")) {
+ containsPort = true;
+ port = targetHost.substring(targetHost.indexOf(':') + 1, targetHost.length());
+ targetHost = targetHost.substring(0, targetHost.indexOf(':'));
+ }
+ //Gathering required information for domain mapping done
+
+ boolean isValidHost = tlbMembershipHandler.isAValidHostName(targetHost);
+ DomainMapping domainMapping = null;
+ if(!isValidHost){
+ //check if the host is valid, if not valid, execute following code to check whether it is a mapped domain
+ domainMapping = mappingCache.getMapping(targetHost);
+ if(domainMapping == null){
+ registryManager = new RegistryManager();
+ domainMapping = registryManager.getMapping(targetHost);
+ mappingCache.addValidMapping(targetHost, domainMapping);
+ }
+ if (domainMapping != null) {
+ actualHost = domainMapping.getActualHost();
+
+ if(containsPort){
+ transportHeaders.put(HTTP.TARGET_HOST, actualHost + ":" + port);
+ } else {
+ transportHeaders.put(HTTP.TARGET_HOST, actualHost);
+ }
+ ((Axis2MessageContext) synCtx).getAxis2MessageContext().setProperty("TRANSPORT_HEADERS" , transportHeaders);
+
+ } else {
+ String msg = "Invalid host name : " + targetHost;
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ }
+
+ if (isSessionAffinityBasedLB()) {
+ // first check if this session is associated with a session. if so, get the endpoint
+ // associated for that session.
+ sessionInformation =
+ (SessionInformation) synCtx.getProperty(
+ SynapseConstants.PROP_SAL_CURRENT_SESSION_INFORMATION);
+
+ currentMember = (Member) synCtx.getProperty(
+ SynapseConstants.PROP_SAL_ENDPOINT_CURRENT_MEMBER);
+
+ if (sessionInformation == null && currentMember == null) {
+ sessionInformation = dispatcher.getSession(synCtx);
+ if (sessionInformation != null) {
+
+ if (log.isDebugEnabled()) {
+ log.debug("Current session id : " + sessionInformation.getId());
+ }
+
+ currentMember = sessionInformation.getMember();
+ synCtx.setProperty(
+ SynapseConstants.PROP_SAL_ENDPOINT_CURRENT_MEMBER, currentMember);
+ // This is for reliably recovery any session information if while response is getting ,
+ // session information has been removed by cleaner.
+ // This will not be a cost as session information a not heavy data structure
+ synCtx.setProperty(
+ SynapseConstants.PROP_SAL_CURRENT_SESSION_INFORMATION, sessionInformation);
+ }
+ }
+
+ }
+
+ // Dispatch request the relevant member
+// String targetHost = getTargetHost(synCtx);
+ ConfigurationContext configCtx =
+ ((Axis2MessageContext) synCtx).getAxis2MessageContext().getConfigurationContext();
+
+ if (tlbMembershipHandler.getConfigurationContext() == null) {
+ tlbMembershipHandler.setConfigurationContext(configCtx);
+ }
+
+ if(tlbMembershipHandler.getClusteringAgent() == null) {
+ tlbMembershipHandler.setConfigurationContext(configCtx);
+ }
+
+ TenantDynamicLoadBalanceFaultHandlerImpl faultHandler = new TenantDynamicLoadBalanceFaultHandlerImpl();
+ log.debug("************* Actual Host: "+actualHost +" ****** Target Host: "+targetHost);
+ faultHandler.setHost(actualHost != null ? actualHost : targetHost);
+
+ if (sessionInformation != null && currentMember != null) {
+ //send message on current session
+ sessionInformation.updateExpiryTime();
+ sendToApplicationMember(synCtx, currentMember, faultHandler, false);
+ } else {
+// prepare for a new session
+ int tenantId = getTenantId(synCtx);
+ //check if this is a valid host name registered in ELB
+ if(tlbMembershipHandler.isAValidHostName(targetHost)){
+ currentMember = tlbMembershipHandler.getNextApplicationMember(targetHost, tenantId);
+ if (currentMember == null) {
+ String msg = "No application members available";
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ sendToApplicationMember(synCtx, currentMember, faultHandler, true);
+ } else {
+ if(domainMapping == null){
+ registryManager = new RegistryManager();
+ domainMapping = registryManager.getMapping(targetHost);
+ mappingCache.addValidMapping(targetHost, domainMapping);
+ }
+ if(domainMapping != null) {
+
+ actualHost = domainMapping.getActualHost();
+
+ log.debug("************* Actual Host: "+actualHost +" ****** Target Host: "+targetHost);
+ faultHandler.setHost(actualHost != null ? actualHost : targetHost);
+
+ if(containsPort){
+ transportHeaders.put(HTTP.TARGET_HOST, actualHost + ":" + port);
+ } else {
+ transportHeaders.put(HTTP.TARGET_HOST, actualHost);
+ }
+ ((Axis2MessageContext) synCtx).getAxis2MessageContext().setProperty("TRANSPORT_HEADERS" , transportHeaders);
+
+ currentMember = tlbMembershipHandler.getNextApplicationMember(actualHost,tenantId);
+ sendToApplicationMember(synCtx,currentMember,faultHandler,true);
+ }else {
+ String msg = "Invalid host name : " + targetHost;
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ }
+ }
+ }
+
+
+// public List<HostContext> getHostContexts() {
+// return Collections.unmodifiableList(hostContexts);
+// }
+
+ /**
+ * This FaultHandler will try to resend the message to another member if an error occurs
+ * while sending to some member. This is a failover mechanism
+ */
+
+ /**
+ * @param url to url for target
+ * @return tenantID if tenant id available else 0
+ */
+ private int getTenantId(String url) {
+ String servicesPrefix = "/t/";
+ if (url != null && url.contains(servicesPrefix)) {
+ int domainNameStartIndex =
+ url.indexOf(servicesPrefix) + servicesPrefix.length();
+ int domainNameEndIndex = url.indexOf('/', domainNameStartIndex);
+ String domainName = url.substring(domainNameStartIndex,
+ domainNameEndIndex == -1 ? url.length() : domainNameEndIndex);
+
+ // return tenant id if domain name is not null
+ if (domainName != null) {
+ try {
+ return ConfigHolder.getInstance().getRealmService().getTenantManager().getTenantId(domainName);
+ } catch (org.wso2.carbon.user.api.UserStoreException e) {
+ log.error("An error occurred while obtaining the tenant id.", e);
+ }
+ }
+ }
+ // return 0 if the domain name is null
+ return 0;
+ }
+
+ private int getTenantId(MessageContext synCtx){
+ String url = synCtx.getTo().toString();
+ int tenantId = getTenantId(url);
+ // tenantId = 0 because domain name was null. May be this is the SSO response
+ if(tenantId == 0 && url.contains(MultitenantConstants.TENANT_DOMAIN+"=")){
+ // OK,this is the SAML SSO response from the IS
+ // e.g url = https://localhost:9444/acs?teantDomain=domain
+ String domainName = url.split("=").clone()[1];
+ // return tenant id if domain name is not null
+ if (domainName != null) {
+ try {
+ return ConfigHolder.getInstance().getRealmService().getTenantManager().getTenantId(domainName);
+ } catch (org.wso2.carbon.user.api.UserStoreException e) {
+ log.error("An error occurred while obtaining the tenant id.", e);
+ }
+ }
+ }
+ return tenantId;
+ }
+
+
+ /**
+ * This FaultHandler will try to resend the message to another member if an error occurs
+ * while sending to some member. This is a failover mechanism
+ */
+ private class TenantDynamicLoadBalanceFaultHandlerImpl extends DynamicLoadbalanceFaultHandler {
+
+ private EndpointReference to;
+ private Member currentMember;
+ private Endpoint currentEp;
+ private String host;
+
+ private static final int MAX_RETRY_COUNT = 5;
+
+ // ThreadLocal variable to keep track of how many times this fault handler has been
+ // called
+ private ThreadLocal<Integer> callCount = new ThreadLocal<Integer>() {
+ protected Integer initialValue() {
+ return 0;
+ }
+ };
+
+ public void setHost(String host) {
+ log.debug("Setting host name: "+host);
+ this.host = host;
+ }
+
+ public void setCurrentMember(Member currentMember) {
+ this.currentMember = currentMember;
+ }
+
+ public void setTo(EndpointReference to) {
+ this.to = to;
+ }
+
+ private TenantDynamicLoadBalanceFaultHandlerImpl() {
+ }
+
+ public void onFault(MessageContext synCtx) {
+ if (currentMember == null || to == null) {
+ return;
+ }
+
+ // Prevent infinite retrying to failed members
+ callCount.set(callCount.get() + 1);
+ if (callCount.get() >= MAX_RETRY_COUNT) {
+ log.debug("Retrying to a failed member has stopped.");
+ return;
+ }
+
+ //cleanup endpoint if exists
+ if (currentEp != null) {
+ currentEp.destroy();
+ }
+ Integer errorCode = (Integer) synCtx.getProperty(SynapseConstants.ERROR_CODE);
+ if (errorCode != null) {
+ if (errorCode.equals(NhttpConstants.CONNECTION_FAILED)) {
+ currentMember.suspend(10000); // TODO: Make this configurable.
+ log.info("Suspended member " + currentMember + " for 10s due to connection failure to that member");
+ }
+ if (errorCode.equals(NhttpConstants.CONNECTION_FAILED) ||
+ errorCode.equals(NhttpConstants.CONNECT_CANCEL) ||
+ errorCode.equals(NhttpConstants.CONNECT_TIMEOUT)) {
+
+ if (!synCtx.getFaultStack().isEmpty()) {
+ synCtx.getFaultStack().pop();
+ }
+ // Try to resend to another member
+ Member newMember = tlbMembershipHandler.getNextApplicationMember(host, getTenantId(synCtx.toString()));
+ if (newMember == null || newMember.isSuspended()) {
+ String msg = "No application members available having host name : "+host+
+ " and tenant id : "+getTenantId(synCtx.toString()+" and which is not suspended.");
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ synCtx.setTo(to);
+ if (isSessionAffinityBasedLB()) {
+ // We are sending this message on a new session,
+ // hence we need to remove previous session information
+ Set pros = synCtx.getPropertyKeySet();
+ if (pros != null) {
+ pros.remove(SynapseConstants.PROP_SAL_CURRENT_SESSION_INFORMATION);
+ }
+ }
+ try {
+ Thread.sleep(1000); // Sleep for sometime before retrying
+ } catch (InterruptedException ignored) {
+ }
+
+ if(synCtx == null || to == null) {
+ return;
+ }
+ log.info("Failed over to " + newMember);
+ sendToApplicationMember(synCtx, newMember, this, true);
+ } else if (errorCode.equals(NhttpConstants.SND_IO_ERROR_SENDING) ||
+ errorCode.equals(NhttpConstants.CONNECTION_CLOSED)) {
+ // TODO: Envelope is consumed
+ String msg = "Error sending request! Connection to host "+host+
+ " might be closed. Error code: "+errorCode;
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ }
+ // We cannot failover since we are using binary relay
+ }
+
+ public void setCurrentEp(Endpoint currentEp) {
+ this.currentEp = currentEp;
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/c1efa393/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/group/mgt/GroupMgtAgentBuilder.java
----------------------------------------------------------------------
diff --git a/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/group/mgt/GroupMgtAgentBuilder.java b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/group/mgt/GroupMgtAgentBuilder.java
new file mode 100644
index 0000000..da25156
--- /dev/null
+++ b/components/load-balancer/lb-endpoint/org.wso2.carbon.lb.endpoint/4.1.3/src/main/java/org/wso2/carbon/lb/endpoint/group/mgt/GroupMgtAgentBuilder.java
@@ -0,0 +1,62 @@
+package org.wso2.carbon.lb.endpoint.group.mgt;
+
+import org.apache.axis2.clustering.ClusteringAgent;
+import org.apache.axis2.clustering.management.GroupManagementAgent;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.synapse.SynapseException;
+import org.wso2.carbon.lb.common.group.mgt.SubDomainAwareGroupManagementAgent;
+import org.wso2.carbon.lb.endpoint.util.ConfigHolder;
+
+/**
+ * Responsible for building {@link GroupManagementAgent}s.
+ */
+public class GroupMgtAgentBuilder {
+
+ private static final Log log = LogFactory.getLog(GroupMgtAgentBuilder.class);
+
+ /**
+ * Creates a {@link SubDomainAwareGroupManagementAgent} corresponds to the given
+ * parameters, if and only if there's no existing agent.
+ * @param domain clustering domain.
+ * @param subDomain clustering sub domain.
+ */
+ public static void createGroupMgtAgent(String domain, String subDomain) {
+
+ ClusteringAgent clusteringAgent =
+ ConfigHolder.getInstance().getAxisConfiguration().getClusteringAgent();
+
+ if (clusteringAgent == null) {
+ throw new SynapseException("Axis2 Clustering Agent not defined in axis2.xml");
+ }
+
+ // checks the existence.
+ if (clusteringAgent.getGroupManagementAgent(domain, subDomain) == null) {
+
+ clusteringAgent.addGroupManagementAgent(new SubDomainAwareGroupManagementAgent(subDomain),
+ domain, subDomain,-1);
+
+ log.info("Group management agent added to cluster domain: " +
+ domain + " and sub domain: " + subDomain);
+ }
+ }
+
+ public static void resetGroupMgtAgent(String domain, String subDomain) {
+
+ ClusteringAgent clusteringAgent =
+ ConfigHolder.getInstance().getAxisConfiguration().getClusteringAgent();
+
+ if (clusteringAgent == null) {
+ throw new SynapseException("Axis2 Clustering Agent not defined in axis2.xml");
+ }
+
+ // checks the existence.
+ if (clusteringAgent.getGroupManagementAgent(domain, subDomain) != null) {
+
+ clusteringAgent.resetGroupManagementAgent(domain, subDomain);
+
+ log.info("Group management agent of cluster domain: " +
+ domain + " and sub domain: " + subDomain+" is removed.");
+ }
+ }
+}