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/03/03 14:14:19 UTC
[1/3] stratos git commit: Introducing load balancer topology provider
model to be able to provide the topology required for load balancing in a
generic manner
Repository: stratos
Updated Branches:
refs/heads/master d76559436 -> c799abceb
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTenantEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTenantEventReceiver.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTenantEventReceiver.java
deleted file mode 100644
index 899822a..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTenantEventReceiver.java
+++ /dev/null
@@ -1,148 +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.messaging.receiver;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.load.balancer.context.LoadBalancerContextUtil;
-import org.apache.stratos.messaging.domain.tenant.Subscription;
-import org.apache.stratos.messaging.domain.tenant.Tenant;
-import org.apache.stratos.messaging.domain.topology.Service;
-import org.apache.stratos.messaging.domain.topology.ServiceType;
-import org.apache.stratos.messaging.event.Event;
-import org.apache.stratos.messaging.event.tenant.CompleteTenantEvent;
-import org.apache.stratos.messaging.event.tenant.TenantSubscribedEvent;
-import org.apache.stratos.messaging.event.tenant.TenantUnSubscribedEvent;
-import org.apache.stratos.messaging.listener.tenant.CompleteTenantEventListener;
-import org.apache.stratos.messaging.listener.tenant.TenantSubscribedEventListener;
-import org.apache.stratos.messaging.listener.tenant.TenantUnSubscribedEventListener;
-import org.apache.stratos.messaging.message.receiver.tenant.TenantEventReceiver;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
-
-/**
- * Load balancer tenant receiver updates load balancer context according to
- * incoming tenant events.
- */
-public class LoadBalancerTenantEventReceiver extends TenantEventReceiver {
-
- private static final Log log = LogFactory.getLog(LoadBalancerTenantEventReceiver.class);
-
- public LoadBalancerTenantEventReceiver() {
- addEventListeners();
- }
-
- private void addEventListeners() {
- addEventListener(new CompleteTenantEventListener() {
- private boolean initialized;
-
- @Override
- protected void onEvent(Event event) {
- if (!initialized) {
- CompleteTenantEvent completeTenantEvent = (CompleteTenantEvent) event;
- if (log.isDebugEnabled()) {
- log.debug("Complete tenant event received");
- }
- for (Tenant tenant : completeTenantEvent.getTenants()) {
- for (Subscription subscription : tenant.getSubscriptions()) {
- if (isMultiTenantService(subscription.getServiceName())) {
- LoadBalancerContextUtil.addClustersAgainstHostNamesAndTenantIds(
- subscription.getServiceName(),
- tenant.getTenantId(),
- subscription.getClusterIds());
- }
- }
- }
- initialized = true;
- }
- }
- });
-
- addEventListener(new TenantSubscribedEventListener() {
- @Override
- protected void onEvent(Event event) {
- TenantSubscribedEvent tenantSubscribedEvent = (TenantSubscribedEvent) event;
- if (log.isDebugEnabled()) {
- log.debug(String.format("Tenant subscribed event received: [tenant-id] %d [service] %s [cluster-ids] %s",
- tenantSubscribedEvent.getTenantId(),
- tenantSubscribedEvent.getServiceName(),
- tenantSubscribedEvent.getClusterIds()));
- }
-
- if (isMultiTenantService(tenantSubscribedEvent.getServiceName())) {
- LoadBalancerContextUtil.addClustersAgainstHostNamesAndTenantIds(
- tenantSubscribedEvent.getServiceName(),
- tenantSubscribedEvent.getTenantId(),
- tenantSubscribedEvent.getClusterIds());
- }
- }
- });
-
- addEventListener(new TenantUnSubscribedEventListener() {
- @Override
- protected void onEvent(Event event) {
- TenantUnSubscribedEvent tenantUnSubscribedEvent = (TenantUnSubscribedEvent) event;
- if (log.isDebugEnabled()) {
- log.debug(String.format("Tenant un-subscribed event received: [tenant-id] %d [service] %s [cluster-ids] %s",
- tenantUnSubscribedEvent.getTenantId(),
- tenantUnSubscribedEvent.getServiceName(),
- tenantUnSubscribedEvent.getClusterIds()));
- }
-
- if (isMultiTenantService(tenantUnSubscribedEvent.getServiceName())) {
- LoadBalancerContextUtil.removeClustersAgainstHostNamesAndTenantIds(
- tenantUnSubscribedEvent.getServiceName(),
- tenantUnSubscribedEvent.getTenantId(),
- tenantUnSubscribedEvent.getClusterIds()
- );
- }
-
- LoadBalancerContextUtil.removeClustersAgainstAllDomains(
- tenantUnSubscribedEvent.getServiceName(),
- tenantUnSubscribedEvent.getTenantId(),
- tenantUnSubscribedEvent.getClusterIds());
-
- LoadBalancerContextUtil.removeAppContextAgainstAllDomains(
- tenantUnSubscribedEvent.getServiceName(),
- tenantUnSubscribedEvent.getTenantId());
- }
- });
- }
-
- private boolean isMultiTenantService(String serviceName) {
- try {
- TopologyManager.acquireReadLock();
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service != null) {
- return (service.getServiceType() == ServiceType.MultiTenant);
- }
- return false;
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- public void execute() {
- super.execute();
-
- if (log.isInfoEnabled()) {
- log.info("Load balancer tenant receiver thread terminated");
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/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
deleted file mode 100644
index 9e036bf..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerTopologyEventReceiver.java
+++ /dev/null
@@ -1,396 +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.messaging.receiver;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.load.balancer.context.LoadBalancerContext;
-import org.apache.stratos.load.balancer.context.LoadBalancerContextUtil;
-import org.apache.stratos.load.balancer.context.map.AlgorithmContextMap;
-import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.Member;
-import org.apache.stratos.messaging.domain.topology.MemberStatus;
-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;
-
-/**
- * Load balancer topology receiver updates load balancer context according to
- * incoming topology events.
- */
-public class LoadBalancerTopologyEventReceiver extends TopologyEventReceiver {
-
- private static final Log log = LogFactory.getLog(LoadBalancerTopologyEventReceiver.class);
-
- public LoadBalancerTopologyEventReceiver() {
- addEventListeners();
- }
-
- public void execute() {
- super.execute();
- if (log.isInfoEnabled()) {
- log.info("Load balancer topology receiver thread started");
- }
- }
-
- private void addEventListeners() {
- // Listen to topology events that affect clusters
- addEventListener(new CompleteTopologyEventListener() {
- private boolean initialized;
-
- @Override
- protected void onEvent(Event event) {
- if (!initialized) {
- try {
- TopologyManager.acquireReadLock();
- for (Service service : TopologyManager.getTopology().getServices()) {
- for (Cluster cluster : service.getClusters()) {
- if (clusterHasActiveMembers(cluster)) {
- LoadBalancerContextUtil.addClusterAgainstHostNames(cluster);
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Cluster does not have any active members");
- }
- }
- for (Member member : cluster.getMembers()) {
- if (member.getStatus() == MemberStatus.Active) {
- addMemberIpsToMemberIpHostnameMap(cluster, member);
- }
-
- }
- }
- }
- initialized = true;
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
- }
-
- private boolean clusterHasActiveMembers(Cluster cluster) {
- for (Member member : cluster.getMembers()) {
- if (member.getStatus() == MemberStatus.Active) {
- return true;
- }
- }
- return false;
- }
- });
- addEventListener(new MemberActivatedEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- MemberActivatedEvent memberActivatedEvent = (MemberActivatedEvent) event;
-
- //TopologyManager.acquireReadLock();
- TopologyManager.acquireReadLockForCluster(memberActivatedEvent.getServiceName(),
- memberActivatedEvent.getClusterId());
-
- try {
-
- Service service = TopologyManager.getTopology().getService(memberActivatedEvent.getServiceName());
- if (service == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Service not found in topology: [service] %s",
- memberActivatedEvent.getServiceName()));
- }
- return;
- }
- Cluster cluster = service.getCluster(memberActivatedEvent.getClusterId());
- if (cluster == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Cluster not found in topology: [service] %s [cluster] %s",
- memberActivatedEvent.getServiceName(), memberActivatedEvent.getClusterId()));
- }
- return;
- }
- Member member = cluster.getMember(memberActivatedEvent.getMemberId());
- if (member == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Member not found in topology: [service] %s [cluster] %s [member] %s",
- memberActivatedEvent.getServiceName(), memberActivatedEvent.getClusterId(),
- memberActivatedEvent.getMemberId()));
- }
- return;
- }
-
- // Add member to member-ip -> hostname map
- addMemberIpsToMemberIpHostnameMap(cluster, member);
-
- if (LoadBalancerContext.getInstance().getClusterIdClusterMap().containsCluster(
- member.getClusterId())) {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster already exists in load balancer context: [service] %s " +
- "[cluster] %s", member.getServiceName(), member.getClusterId()));
- }
- // At this point member is already added to the cluster object in load balancer context
- return;
- }
-
- // Add cluster to load balancer context when its first member is activated:
- // Cluster not found in load balancer context, add it
- LoadBalancerContextUtil.addClusterAgainstHostNames(cluster);
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- //TopologyManager.releaseReadLock();
- TopologyManager.releaseReadLockForCluster(memberActivatedEvent.getServiceName(),
- memberActivatedEvent.getClusterId());
- }
- }
- });
- addEventListener(new MemberMaintenanceListener() {
- @Override
- protected void onEvent(Event event) {
-
- MemberMaintenanceModeEvent memberMaintenanceModeEvent = (MemberMaintenanceModeEvent) event;
-
- TopologyManager.acquireReadLockForCluster(memberMaintenanceModeEvent.getServiceName(),
- memberMaintenanceModeEvent.getClusterId());
-
- try {
- //TopologyManager.acquireReadLock();
-
- Member member = findMember(memberMaintenanceModeEvent.getServiceName(),
- memberMaintenanceModeEvent.getClusterId(), memberMaintenanceModeEvent.getMemberId());
-
- if (member != null) {
- removeMemberIpsFromMemberIpHostnameMap(member);
- }
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- //TopologyManager.releaseReadLock();
- TopologyManager.releaseReadLockForCluster(memberMaintenanceModeEvent.getServiceName(),
- memberMaintenanceModeEvent.getClusterId());
- }
- }
- });
- addEventListener(new MemberSuspendedEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- MemberSuspendedEvent memberSuspendedEvent = (MemberSuspendedEvent) event;
- TopologyManager.acquireReadLockForCluster(memberSuspendedEvent.getServiceName(),
- memberSuspendedEvent.getClusterId());
-
- try {
- //TopologyManager.acquireReadLock();
- Member member = findMember(memberSuspendedEvent.getServiceName(),
- memberSuspendedEvent.getClusterId(), memberSuspendedEvent.getMemberId());
-
- if (member != null) {
- removeMemberIpsFromMemberIpHostnameMap(member);
- }
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- //TopologyManager.releaseReadLock();
- TopologyManager.releaseReadLockForCluster(memberSuspendedEvent.getServiceName(),
- memberSuspendedEvent.getClusterId());
- }
- }
- });
- addEventListener(new MemberTerminatedEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- //TopologyManager.acquireReadLock();
- MemberTerminatedEvent memberTerminatedEvent = (MemberTerminatedEvent) event;
-
- TopologyManager.acquireReadLockForCluster(memberTerminatedEvent.getServiceName(),
- memberTerminatedEvent.getClusterId());
-
- try {
- Member member = findMember(memberTerminatedEvent.getServiceName(),
- memberTerminatedEvent.getClusterId(), memberTerminatedEvent.getMemberId());
-
- if (member != null) {
- removeMemberIpsFromMemberIpHostnameMap(member);
- }
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- //TopologyManager.releaseReadLock();
- TopologyManager.releaseReadLockForCluster(memberTerminatedEvent.getServiceName(),
- memberTerminatedEvent.getClusterId());
- }
- }
- });
- addEventListener(new ClusterRemovedEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- // Remove cluster from context
- ClusterRemovedEvent clusterRemovedEvent = (ClusterRemovedEvent) event;
- TopologyManager.acquireReadLockForCluster(clusterRemovedEvent.getServiceName(),
- clusterRemovedEvent.getClusterId());
-
- try {
- AlgorithmContextMap.getInstance().removeCluster(clusterRemovedEvent.getServiceName(),
- clusterRemovedEvent.getClusterId());
- } catch (Exception e) {
- log.error("Could not remove cluster from load balancer algorithm context map", e);
- }
-
- try {
- Cluster cluster = LoadBalancerContext.getInstance().getClusterIdClusterMap().getCluster(clusterRemovedEvent.getClusterId());
- if (cluster != null) {
- for (Member member : cluster.getMembers()) {
- removeMemberIpsFromMemberIpHostnameMap(member);
- }
- LoadBalancerContextUtil.removeClusterAgainstHostNames(cluster.getClusterId());
- } else {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Cluster not found in load balancer context: [service] %s [cluster] %s",
- clusterRemovedEvent.getServiceName(), clusterRemovedEvent.getClusterId()));
- }
- }
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- TopologyManager.releaseReadLockForCluster(clusterRemovedEvent.getServiceName(),
- clusterRemovedEvent.getClusterId());
- }
- }
- });
- addEventListener(new ServiceRemovedEventListener() {
- @Override
- protected void onEvent(Event event) {
-
- ServiceRemovedEvent serviceRemovedEvent = (ServiceRemovedEvent) event;
- TopologyManager.acquireReadLockForService(serviceRemovedEvent.getServiceName());
-
- try {
- //TopologyManager.acquireReadLock();
-
- // Remove all clusters of given service from context
- Service service = TopologyManager.getTopology().getService(serviceRemovedEvent.getServiceName());
- if (service != null) {
- for (Cluster cluster : service.getClusters()) {
- for (Member member : cluster.getMembers()) {
- removeMemberIpsFromMemberIpHostnameMap(member);
- }
- LoadBalancerContextUtil.removeClusterAgainstHostNames(cluster.getClusterId());
- }
- } else {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Service not found in topology: [service] %s",
- serviceRemovedEvent.getServiceName()));
- }
- }
- } catch (Exception e) {
- log.error("Error processing event", e);
- } finally {
- //TopologyManager.releaseReadLock();
- TopologyManager.releaseReadLockForService(serviceRemovedEvent.getServiceName());
- }
- }
- });
- }
-
- private Member findMember(String serviceName, String clusterId, String memberId) {
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Service not found in topology: [service] %s", serviceName));
- }
- return null;
- }
-
- Cluster cluster = service.getCluster(clusterId);
- if (cluster == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Cluster not found in topology: [service] %s [cluster] %s", serviceName, clusterId));
- }
- return null;
- }
-
- Member member = cluster.getMember(memberId);
- if (member == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Member not found in topology: [service] %s [cluster] %s [member] %s", serviceName,
- clusterId, memberId));
- }
- return null;
- }
- return member;
- }
-
- private void addMemberIpsToMemberIpHostnameMap(Cluster cluster, Member member) {
- if ((cluster.getHostNames() == null) || (cluster.getHostNames().size() == 0)) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Hostnames not found in cluster %s, could not add member ips to member-ip " +
- "-> hostname map", member.getClusterId()));
- }
- return;
- }
-
- String hostname = cluster.getHostNames().get(0);
- if (cluster.getHostNames().size() > 1) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Multiple hostnames found in cluster %s, using %s",
- cluster.getHostNames().toString(), hostname));
- }
- }
-
- if (StringUtils.isNotBlank(member.getDefaultPrivateIP())) {
- LoadBalancerContext.getInstance().getMemberIpHostnameMap().put(member.getDefaultPrivateIP(), hostname);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Member private ip added to member-ip -> hostname map: [service] %s [cluster] " +
- "%s [member] %s [private-ip] %s", member.getServiceName(), member.getClusterId(),
- member.getMemberId(), member.getDefaultPrivateIP()
- ));
- }
- }
- if (StringUtils.isNotBlank(member.getDefaultPublicIP())) {
- LoadBalancerContext.getInstance().getMemberIpHostnameMap().put(member.getDefaultPublicIP(), hostname);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Member public ip added to member-ip -> hostname map: [service] %s [cluster] " +
- "%s [member] %s [public-ip] %s", member.getServiceName(), member.getClusterId(),
- member.getMemberId(), member.getDefaultPublicIP()
- ));
- }
- }
- }
-
- private void removeMemberIpsFromMemberIpHostnameMap(Member member) {
- if (StringUtils.isNotBlank(member.getDefaultPrivateIP())) {
- LoadBalancerContext.getInstance().getMemberIpHostnameMap().remove(member.getDefaultPrivateIP());
- if (log.isDebugEnabled()) {
- log.debug(String.format("Member private ip removed from member-ip -> hostname map: [private-ip] %s",
- member.getDefaultPrivateIP()));
- }
- }
- if (StringUtils.isNotBlank(member.getDefaultPublicIP())) {
- LoadBalancerContext.getInstance().getMemberIpHostnameMap().remove(member.getDefaultPublicIP());
- if (log.isDebugEnabled()) {
- log.debug(String.format("Member public ip removed from member-ip -> hostname map: [public-ip] %s",
- member.getDefaultPublicIP()));
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsCollector.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsCollector.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsCollector.java
index 10e38f9..a5239b4 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsCollector.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/statistics/LoadBalancerStatisticsCollector.java
@@ -22,10 +22,8 @@ import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.load.balancer.common.statistics.LoadBalancerStatisticsReader;
-import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.Member;
-import java.util.*;
+import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -95,15 +93,8 @@ public class LoadBalancerStatisticsCollector implements LoadBalancerStatisticsRe
}
}
- public int getActiveInstancesCount(Cluster cluster) {
- int activeInstances = 0;
- for( Member member :cluster.getMembers()){
- if(member.isActive()){
- activeInstances++;
- }
-
- }return activeInstances;
-
+ public int getActiveInstancesCount(org.apache.stratos.messaging.domain.topology.Cluster cluster) {
+ return cluster.getMembers().size();
}
void incrementInFlightRequestCount(String clusterId) {
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/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
index 2ddc7fa..60e01d9 100644
--- 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
@@ -37,4 +37,5 @@ public class LoadBalancerConstants {
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;
+ public static final String CLUSTER_HOSTNAME = "CLUSTER_HOSTNAME";
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/test/java/org/apache/stratos/load/balancer/test/LoadBalancerConfigurationTest.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/test/java/org/apache/stratos/load/balancer/test/LoadBalancerConfigurationTest.java b/components/org.apache.stratos.load.balancer/src/test/java/org/apache/stratos/load/balancer/test/LoadBalancerConfigurationTest.java
index d17a270..d8ee1fd 100755
--- a/components/org.apache.stratos.load.balancer/src/test/java/org/apache/stratos/load/balancer/test/LoadBalancerConfigurationTest.java
+++ b/components/org.apache.stratos.load.balancer/src/test/java/org/apache/stratos/load/balancer/test/LoadBalancerConfigurationTest.java
@@ -18,10 +18,12 @@
*/
package org.apache.stratos.load.balancer.test;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
+import org.apache.stratos.load.balancer.common.domain.Cluster;
+import org.apache.stratos.load.balancer.common.domain.Member;
+import org.apache.stratos.load.balancer.common.domain.Port;
import org.apache.stratos.load.balancer.conf.LoadBalancerConfiguration;
import org.apache.stratos.load.balancer.conf.domain.TenantIdentifier;
-import org.apache.stratos.messaging.domain.topology.*;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -101,18 +103,10 @@ public class LoadBalancerConfigurationTest {
try {
String validationError = "Static topology validation failed";
- TopologyManager.acquireReadLock();
- Topology topology = TopologyManager.getTopology();
- Assert.assertTrue(String.format("%s, services not found", validationError), topology.getServices().size() > 0);
-
- String serviceName = "app-server";
- Service appServer = topology.getService(serviceName);
- Assert.assertNotNull(String.format("%s, service not found: [service] %s", validationError, serviceName), appServer);
-
- Assert.assertTrue(String.format("%s, multi-tenant is not true: [service] %s", validationError, serviceName), appServer.getServiceType() == ServiceType.MultiTenant);
+ TopologyProvider topologyProvider = LoadBalancerConfiguration.getInstance().getTopologyProvider();
String clusterId = "app-server-cluster1";
- Cluster cluster1 = appServer.getCluster(clusterId);
+ Cluster cluster1 = topologyProvider.getClusterByClusterId(clusterId);
Assert.assertNotNull(String.format("%s, cluster not found: [cluster] %s", validationError, clusterId), cluster1);
Assert.assertEquals(String.format("%s, tenant range is not valid: [cluster] %s", validationError, clusterId), cluster1.getTenantRange(), "1-100");
@@ -126,7 +120,7 @@ public class LoadBalancerConfigurationTest {
String memberId = "m1";
Member m1 = cluster1.getMember(memberId);
Assert.assertNotNull(String.format("%s, member not found: [member] %s", validationError, memberId), m1);
- Assert.assertEquals(String.format("%s, member ip not valid", validationError), "10.0.0.10", m1.getDefaultPrivateIP());
+ Assert.assertEquals(String.format("%s, member ip not valid", validationError), "10.0.0.10", m1.getHostName());
int proxyPort = 80;
Port m1Http = m1.getPort(proxyPort);
@@ -138,7 +132,6 @@ public class LoadBalancerConfigurationTest {
Assert.assertTrue(String.format("%s, map-domain-names is not true", validationError), LoadBalancerConfiguration.getInstance().isDomainMappingEnabled());
} finally {
- TopologyManager.releaseReadLock();
LoadBalancerConfiguration.clear();
}
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.rest.endpoint/src/main/java/org/apache/stratos/rest/endpoint/mock/MockContext.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.rest.endpoint/src/main/java/org/apache/stratos/rest/endpoint/mock/MockContext.java b/components/org.apache.stratos.rest.endpoint/src/main/java/org/apache/stratos/rest/endpoint/mock/MockContext.java
index 99de9db..b098823 100644
--- a/components/org.apache.stratos.rest.endpoint/src/main/java/org/apache/stratos/rest/endpoint/mock/MockContext.java
+++ b/components/org.apache.stratos.rest.endpoint/src/main/java/org/apache/stratos/rest/endpoint/mock/MockContext.java
@@ -27,7 +27,6 @@ import org.apache.stratos.common.beans.partition.PartitionBean;
import org.apache.stratos.common.beans.policy.autoscale.AutoscalePolicyBean;
import org.apache.stratos.common.beans.policy.deployment.DeploymentPolicyBean;
import org.apache.stratos.common.beans.topology.ClusterBean;
-import org.apache.stratos.messaging.domain.tenant.SubscriptionDomain;
import org.apache.stratos.rest.endpoint.exception.RestAPIException;
import org.wso2.carbon.context.CarbonContext;
@@ -37,7 +36,6 @@ import java.util.*;
public class MockContext {
private static MockContext mockContext = new MockContext(); // singleton
- private Map<String, List<SubscriptionDomain>> subscriptionAliasToDomainMap = new HashMap<String, List<SubscriptionDomain>>();
private Map<Integer, List<String>> tenantIdToAliasesMap = new HashMap<Integer, List<String>>();
private Map<Integer, List<CartridgeBean>> cartridgeDefinitionBeanList = new HashMap<Integer, List<CartridgeBean>>();
private Map<Integer, Map<String,CartridgeBean>> availableSingleTenantCartridges = new HashMap<Integer, Map<String,CartridgeBean>>();
@@ -739,27 +737,6 @@ public class MockContext {
return (deploymentPolicyMap.get(tenantId)).values().toArray(new DeploymentPolicyBean[0]);
}
}
-
- public ApiResponseBean removeSubscriptionDomain(int tenantId,
- String subscriptionAlias, String domainName) {
- ApiResponseBean stratosApiResponse = new ApiResponseBean();
-
- List<String> tenantAliases = tenantIdToAliasesMap.get(tenantId);
- if(tenantAliases != null && tenantAliases.contains(subscriptionAlias)) {
- for (Iterator<SubscriptionDomain> iterator = subscriptionAliasToDomainMap.get(subscriptionAlias).iterator(); iterator
- .hasNext();) {
- SubscriptionDomain subscriptionDomain = (SubscriptionDomain) iterator.next();
- if (subscriptionDomain.getDomainName().equals(domainName)) {
- iterator.remove();
- stratosApiResponse.setMessage("Successfully removed the subscription domain: "+domainName);
- }
- }
- } else {
- stratosApiResponse.setMessage("Failed to remove the subscription domain: "+domainName);
- }
-
- return stratosApiResponse;
- }
public void addUser(UserInfoBean user) {
int tenantId = getTenantId();
[3/3] stratos git commit: Introducing load balancer topology provider
model to be able to provide the topology required for load balancing in a
generic manner
Posted by im...@apache.org.
Introducing load balancer topology provider model to be able to provide the topology required for load balancing in a generic manner
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/c799abce
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/c799abce
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/c799abce
Branch: refs/heads/master
Commit: c799abceb9e8b0086bc2603cc36377972aec79ae
Parents: d765594
Author: Imesh Gunaratne <im...@apache.org>
Authored: Tue Mar 3 18:43:57 2015 +0530
Committer: Imesh Gunaratne <im...@apache.org>
Committed: Tue Mar 3 18:44:15 2015 +0530
----------------------------------------------------------------------
.../load/balancer/common/domain/Cluster.java | 159 +++++++
.../load/balancer/common/domain/Member.java | 87 ++++
.../load/balancer/common/domain/Port.java | 53 +++
.../load/balancer/common/domain/Service.java | 73 +++
.../load/balancer/common/domain/Topology.java | 56 +++
...dBalancerApplicationSignUpEventReceiver.java | 74 +++
.../LoadBalancerDomainMappingEventReceiver.java | 86 ++++
.../LoadBalancerTopologyEventReceiver.java | 354 ++++++++++++++
.../LoadBalancerStatisticsReader.java | 11 +-
.../common/topology/TopologyProvider.java | 175 +++++++
.../algorithm/LoadBalanceAlgorithm.java | 2 +-
.../load/balancer/algorithm/RoundRobin.java | 4 +-
.../conf/LoadBalancerConfiguration.java | 80 ++--
.../balancer/context/LoadBalancerContext.java | 51 +-
.../context/LoadBalancerContextUtil.java | 475 -------------------
.../context/map/ClusterIdClusterMap.java | 57 ---
.../context/map/HostNameAppContextMap.java | 4 +
.../context/map/HostNameClusterMap.java | 63 ---
.../context/map/MultiTenantClusterMap.java | 77 ---
.../map/ServiceNameServiceContextMap.java | 58 ---
.../TenantIdSynapseEnvironmentServiceMap.java | 63 ---
.../balancer/endpoint/RequestDelegator.java | 48 +-
.../TenantAwareLoadBalanceEndpoint.java | 135 +++---
.../internal/LoadBalancerServiceComponent.java | 118 ++---
.../internal/ServiceReferenceHolder.java | 26 +
.../balancer/mediators/LocationReWriter.java | 5 +-
...dBalancerApplicationSignUpEventReceiver.java | 69 ---
.../LoadBalancerDomainMappingEventReceiver.java | 72 ---
.../LoadBalancerTenantEventReceiver.java | 148 ------
.../LoadBalancerTopologyEventReceiver.java | 396 ----------------
.../LoadBalancerStatisticsCollector.java | 15 +-
.../balancer/util/LoadBalancerConstants.java | 1 +
.../test/LoadBalancerConfigurationTest.java | 21 +-
.../stratos/rest/endpoint/mock/MockContext.java | 23 -
34 files changed, 1380 insertions(+), 1759 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Cluster.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Cluster.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Cluster.java
new file mode 100644
index 0000000..7f2fd9c
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Cluster.java
@@ -0,0 +1,159 @@
+/*
+ * 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.common.domain;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Load balancer cluster definition.
+ */
+public class Cluster {
+
+ private static final Log log = LogFactory.getLog(Cluster.class);
+
+ private String serviceName;
+ private String clusterId;
+ private Set<String> hostNames;
+ private String tenantRange;
+ private Map<String, Member> memberMap;
+ private Map<String, String> hostNameToContextPathMap;
+ private String loadBalanceAlgorithmName;
+
+ public Cluster(String serviceName, String clusterId) {
+ this.serviceName = serviceName;
+ this.clusterId = clusterId;
+ this.hostNames = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+ this.memberMap = new HashMap<String, Member>();
+ this.hostNameToContextPathMap = new ConcurrentHashMap<String, String>();
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public String getClusterId() {
+ return clusterId;
+ }
+
+ public Set<String> getHostNames() {
+ return hostNames;
+ }
+
+ public void addHostName(String hostName) {
+ hostNames.add(hostName);
+ }
+
+ public void addHostName(String hostName, String contextPath) {
+ hostNames.add(hostName);
+ hostNameToContextPathMap.put(hostName, contextPath);
+ }
+
+ public void addMember(Member member) {
+ memberMap.put(member.getMemberId(), member);
+ if(log.isInfoEnabled()) {
+ log.info(String.format("Member added to cluster: [cluster] %s [member] %s [hostname] %s [ports] %s",
+ clusterId, member.getMemberId(), member.getHostName(), member.getPorts()));
+ }
+ }
+
+ public void removeMember(String memberId) {
+ Member member = memberMap.get(memberId);
+ if(member == null) {
+ if(log.isWarnEnabled()) {
+ log.warn(String.format("Could not remove member, member not found: [member] %s", memberId));
+ }
+ }
+
+ memberMap.remove(memberId);
+ if(log.isInfoEnabled()) {
+ log.info(String.format("Member removed from cluster: [cluster] %s [member] %s [hostname] %s",
+ clusterId, member.getMemberId(), member.getHostName()));
+ }
+ }
+
+ public Member getMember(String memberId) {
+ return memberMap.get(memberId);
+ }
+
+ public Collection getMembers() {
+ return memberMap.values();
+ }
+
+ public String getTenantRange() {
+ return tenantRange;
+ }
+
+ public void setTenantRange(String tenantRange) {
+ this.tenantRange = tenantRange;
+ }
+
+ /**
+ * Check whether a given tenant id is in tenant range of the cluster.
+ *
+ * @param tenantId
+ * @return
+ */
+ public boolean tenantIdInRange(int tenantId) {
+ if (StringUtils.isEmpty(getTenantRange())) {
+ return false;
+ }
+
+ if ("*".equals(getTenantRange())) {
+ return true;
+ } else {
+ String[] array = getTenantRange().split("-");
+ int tenantStart = Integer.parseInt(array[0]);
+ if (tenantStart <= tenantId) {
+ String tenantEndStr = array[1];
+ if ("*".equals(tenantEndStr)) {
+ return true;
+ } else {
+ int tenantEnd = Integer.parseInt(tenantEndStr);
+ if (tenantId <= tenantEnd) {
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+ }
+
+ public void setLoadBalanceAlgorithmName(String loadBalanceAlgorithmName) {
+ this.loadBalanceAlgorithmName = loadBalanceAlgorithmName;
+ }
+
+ public String getLoadBalanceAlgorithmName() {
+ return loadBalanceAlgorithmName;
+ }
+
+ public String getContextPath(String hostName) {
+ return hostNameToContextPathMap.get(hostName);
+ }
+
+ public void removeHostName(String hostName) {
+ hostNames.remove(hostName);
+ hostNameToContextPathMap.remove(hostName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Member.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Member.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Member.java
new file mode 100644
index 0000000..dae7e8c
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Member.java
@@ -0,0 +1,87 @@
+/*
+ * 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.common.domain;
+
+import java.util.*;
+
+/**
+ * Load balancer member definition.
+ */
+public class Member {
+
+ private String serviceName;
+ private String clusterId;
+ private String memberId;
+ private String hostName;
+ private Map<Integer, Port> portMap;
+
+ public Member(String serviceName, String clusterId, String memberId, String hostName) {
+ this.serviceName = serviceName;
+ this.clusterId = clusterId;
+ this.memberId = memberId;
+ this.hostName = hostName;
+ this.portMap = new HashMap<Integer, Port>();
+ }
+
+ public String getClusterId() {
+ return clusterId;
+ }
+
+ public String getMemberId() {
+ return memberId;
+ }
+
+ public String getHostName() {
+ return hostName;
+ }
+
+ public Port getPort(int proxy) {
+ if(portMap.containsKey(proxy)) {
+ return portMap.get(proxy);
+ }
+ return null;
+ }
+
+ public void addPort(Port port) {
+ this.portMap.put(port.getProxy(), port);
+ }
+
+ public void addPorts(Collection<Port> ports) {
+ for(Port port : ports) {
+ addPort(port);
+ }
+ }
+
+ public void removePort(Port port) {
+ this.portMap.remove(port.getProxy());
+ }
+
+ public boolean portExists(Port port) {
+ return this.portMap.containsKey(port.getProxy());
+ }
+
+ public Collection getPorts() {
+ return portMap.values();
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Port.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Port.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Port.java
new file mode 100644
index 0000000..07d1784
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Port.java
@@ -0,0 +1,53 @@
+/*
+ * 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.common.domain;
+
+/**
+ * Load balancer port definition.
+ */
+public class Port {
+
+ private String protocol;
+ private int value;
+ private int proxy;
+
+ public Port(String protocol, int value, int proxy) {
+ this.protocol = protocol;
+ this.value = value;
+ this.proxy = proxy;
+ }
+
+ public String getProtocol() {
+ return protocol;
+ }
+
+ public int getValue() {
+ return value;
+ }
+
+ public int getProxy() {
+ return proxy;
+ }
+
+ @Override
+ public String toString() {
+ return "Port [protocol=" + protocol + ", value=" + value + ", proxy=" + proxy + "]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Service.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Service.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Service.java
new file mode 100644
index 0000000..579035b
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Service.java
@@ -0,0 +1,73 @@
+/*
+ * 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.common.domain;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Load balancer topology definition.
+ */
+public class Service {
+ private final String serviceName;
+ private final boolean isMultiTenant;
+ // Key: Cluster.clusterId
+ private Map<String, Cluster> clusterIdClusterMap;
+ private boolean multiTenant;
+
+ public Service(String serviceName, boolean isMultiTenant) {
+ this.serviceName = serviceName;
+ this.clusterIdClusterMap = new ConcurrentHashMap<String, Cluster>();
+ this.isMultiTenant = isMultiTenant;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public Collection<Cluster> getClusters() {
+ return clusterIdClusterMap.values();
+ }
+
+ public void addCluster(Cluster cluster) {
+ this.clusterIdClusterMap.put(cluster.getClusterId(), cluster);
+ }
+
+ public void removeCluster(String clusterId) {
+ this.clusterIdClusterMap.remove(clusterId);
+ }
+
+ public boolean clusterExists(String clusterId) {
+ return this.clusterIdClusterMap.containsKey(clusterId);
+ }
+
+ public Cluster getCluster(String clusterId) {
+ return this.clusterIdClusterMap.get(clusterId);
+ }
+
+ public boolean isMultiTenant() {
+ return multiTenant;
+ }
+
+ public void setMultiTenant(boolean multiTenant) {
+ this.multiTenant = multiTenant;
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Topology.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Topology.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Topology.java
new file mode 100644
index 0000000..59b598d
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/domain/Topology.java
@@ -0,0 +1,56 @@
+/*
+ * 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.common.domain;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Load balancer topology definition.
+ */
+public class Topology {
+
+ private Map<String, Service> serviceMap;
+
+ public Topology() {
+ serviceMap = new ConcurrentHashMap<String, Service>();
+ }
+
+ public void addService(Service service) {
+ this.serviceMap.put(service.getServiceName(), service);
+ }
+
+ public void removeService(String serviceName) {
+ this.serviceMap.remove(serviceName);
+ }
+
+ public Service getService(String serviceName) {
+ return this.serviceMap.get(serviceName);
+ }
+
+ public boolean serviceExists(String serviceName) {
+ return this.serviceMap.containsKey(serviceName);
+ }
+
+ public Collection<Service> getServices() {
+ return serviceMap.values();
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerApplicationSignUpEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerApplicationSignUpEventReceiver.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerApplicationSignUpEventReceiver.java
new file mode 100644
index 0000000..1f4e26b
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerApplicationSignUpEventReceiver.java
@@ -0,0 +1,74 @@
+/*
+ * 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.common.event.receivers;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
+import org.apache.stratos.load.balancer.common.domain.Cluster;
+import org.apache.stratos.messaging.domain.application.signup.ApplicationSignUp;
+import org.apache.stratos.messaging.domain.application.signup.DomainMapping;
+import org.apache.stratos.messaging.event.Event;
+import org.apache.stratos.messaging.event.application.signup.CompleteApplicationSignUpsEvent;
+import org.apache.stratos.messaging.listener.application.signup.CompleteApplicationSignUpsEventListener;
+import org.apache.stratos.messaging.message.receiver.application.signup.ApplicationSignUpEventReceiver;
+
+/**
+ * Load balancer application signup event receiver updates the topology in the given topology provider
+ * with the hostnames found in application signup events.
+ */
+public class LoadBalancerApplicationSignUpEventReceiver extends ApplicationSignUpEventReceiver {
+
+ private static final Log log = LogFactory.getLog(LoadBalancerApplicationSignUpEventReceiver.class);
+
+ private TopologyProvider topologyProvider;
+
+ public LoadBalancerApplicationSignUpEventReceiver(TopologyProvider topologyProvider) {
+ this.topologyProvider = topologyProvider;
+ addEventListeners();
+ }
+
+ private void addEventListeners() {
+ addEventListener(new CompleteApplicationSignUpsEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+ if (log.isDebugEnabled()) {
+ log.debug("Complete application signup event received");
+ }
+
+ CompleteApplicationSignUpsEvent completeApplicationSignUpsEvent = (CompleteApplicationSignUpsEvent)event;
+ for(ApplicationSignUp applicationSignUp : completeApplicationSignUpsEvent.getApplicationSignUps()) {
+ if(applicationSignUp.getDomainMappings() != null) {
+ for (DomainMapping domainMapping : applicationSignUp.getDomainMappings()) {
+ if(domainMapping != null) {
+ Cluster cluster = topologyProvider.getClusterByClusterId(domainMapping.getClusterId());
+ if(cluster != null) {
+ cluster.addHostName(domainMapping.getDomainName());
+ log.info(String.format("Domain mapping added: [cluster] %s [domain] %s",
+ cluster.getClusterId(), domainMapping.getDomainName()));
+ }
+ }
+ }
+ }
+ }
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerDomainMappingEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerDomainMappingEventReceiver.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerDomainMappingEventReceiver.java
new file mode 100644
index 0000000..8634abc
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerDomainMappingEventReceiver.java
@@ -0,0 +1,86 @@
+/*
+ * 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.common.event.receivers;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
+import org.apache.stratos.load.balancer.common.domain.Cluster;
+import org.apache.stratos.messaging.event.Event;
+import org.apache.stratos.messaging.event.domain.mapping.DomainMappingAddedEvent;
+import org.apache.stratos.messaging.event.domain.mapping.DomainMappingRemovedEvent;
+import org.apache.stratos.messaging.listener.domain.mapping.DomainMappingAddedEventListener;
+import org.apache.stratos.messaging.listener.domain.mapping.DomainMappingRemovedEventListener;
+import org.apache.stratos.messaging.message.receiver.domain.mapping.DomainMappingEventReceiver;
+
+/**
+ * Load balancer domain mapping event receiver updates the topology in the given topology provider
+ * with the domains found in domain mapping events.
+ */
+public class LoadBalancerDomainMappingEventReceiver extends DomainMappingEventReceiver {
+
+ private static final Log log = LogFactory.getLog(LoadBalancerDomainMappingEventReceiver.class);
+
+ private TopologyProvider topologyProvider;
+
+ public LoadBalancerDomainMappingEventReceiver(TopologyProvider topologyProvider) {
+ this.topologyProvider = topologyProvider;
+ addEventListeners();
+ }
+
+ private void addEventListeners() {
+
+ addEventListener(new DomainMappingAddedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+ DomainMappingAddedEvent domainMappingAddedEvent = (DomainMappingAddedEvent)event;
+
+ String domainName = domainMappingAddedEvent.getDomainName();
+ String contextPath = domainMappingAddedEvent.getContextPath();
+
+ String clusterId = domainMappingAddedEvent.getClusterId();
+ Cluster cluster = topologyProvider.getClusterByClusterId(clusterId);
+ if(cluster == null) {
+ log.warn(String.format("Could not add domain mapping, cluster not found: [cluster] %s", clusterId));
+ }
+
+ cluster.addHostName(domainName, contextPath);
+ log.info(String.format("Domain mapping added: [cluster] %s [domain] %s", clusterId, domainName));
+ }
+ });
+
+ addEventListener(new DomainMappingRemovedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+ DomainMappingRemovedEvent domainMappingRemovedEvent = (DomainMappingRemovedEvent)event;
+
+ String clusterId = domainMappingRemovedEvent.getClusterId();
+ Cluster cluster = topologyProvider.getClusterByClusterId(clusterId);
+ if(cluster == null) {
+ log.warn(String.format("Could not remove domain mapping, cluster not found: [cluster] %s", clusterId));
+ }
+
+ String domainName = domainMappingRemovedEvent.getDomainName();
+ cluster.removeHostName(domainName);
+ log.info(String.format("Domain mapping removed: [cluster] %s [domain] %s", clusterId, domainName));
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerTopologyEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerTopologyEventReceiver.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerTopologyEventReceiver.java
new file mode 100644
index 0000000..ee5b5b3
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/event/receivers/LoadBalancerTopologyEventReceiver.java
@@ -0,0 +1,354 @@
+/*
+ * 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.common.event.receivers;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
+import org.apache.stratos.messaging.domain.topology.Cluster;
+import org.apache.stratos.messaging.domain.topology.Member;
+import org.apache.stratos.messaging.domain.topology.MemberStatus;
+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;
+
+/**
+ * Load balancer topology receiver updates the topology in the given topology provider
+ * according to topology events.
+ */
+public class LoadBalancerTopologyEventReceiver extends TopologyEventReceiver {
+
+ private static final Log log = LogFactory.getLog(LoadBalancerTopologyEventReceiver.class);
+
+ private TopologyProvider topologyProvider;
+
+ public LoadBalancerTopologyEventReceiver(TopologyProvider topologyProvider) {
+ this.topologyProvider = topologyProvider;
+ addEventListeners();
+ }
+
+ public void execute() {
+ super.execute();
+ if (log.isInfoEnabled()) {
+ log.info("Load balancer topology receiver thread started");
+ }
+ }
+
+ private void addEventListeners() {
+
+ addEventListener(new CompleteTopologyEventListener() {
+ private boolean initialized;
+
+ @Override
+ protected void onEvent(Event event) {
+ if (!initialized) {
+ try {
+ TopologyManager.acquireReadLock();
+ for (Service service : TopologyManager.getTopology().getServices()) {
+ for (Cluster cluster : service.getClusters()) {
+ for (Member member : cluster.getMembers()) {
+ if (member.getStatus() == MemberStatus.Active) {
+ addMember(cluster.getServiceName(), member.getClusterId(), member.getMemberId());
+ }
+ }
+ }
+ }
+ initialized = true;
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+ }
+ });
+
+ addEventListener(new MemberActivatedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ MemberActivatedEvent memberActivatedEvent = (MemberActivatedEvent) event;
+
+ String serviceName = memberActivatedEvent.getServiceName();
+ String clusterId = memberActivatedEvent.getClusterId();
+ String memberId = memberActivatedEvent.getMemberId();
+
+ try {
+ TopologyManager.acquireReadLockForCluster(serviceName, clusterId);
+ addMember(serviceName, clusterId, memberId);
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLockForCluster(serviceName, clusterId);
+ }
+ }
+ });
+
+ addEventListener(new MemberMaintenanceListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ MemberMaintenanceModeEvent memberMaintenanceModeEvent = (MemberMaintenanceModeEvent) event;
+
+ String serviceName = memberMaintenanceModeEvent.getServiceName();
+ String clusterId = memberMaintenanceModeEvent.getClusterId();
+ String memberId = memberMaintenanceModeEvent.getMemberId();
+
+ try {
+ TopologyManager.acquireReadLockForCluster(serviceName, clusterId);
+ removeMember(serviceName, clusterId, memberId);
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLockForCluster(serviceName,
+ clusterId);
+ }
+ }
+ });
+
+ addEventListener(new MemberSuspendedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ MemberSuspendedEvent memberSuspendedEvent = (MemberSuspendedEvent) event;
+
+ String serviceName = memberSuspendedEvent.getServiceName();
+ String clusterId = memberSuspendedEvent.getClusterId();
+ String memberId = memberSuspendedEvent.getMemberId();
+
+ try {
+ TopologyManager.acquireReadLockForCluster(serviceName, clusterId);
+ removeMember(serviceName, clusterId, memberId);
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLockForCluster(memberSuspendedEvent.getServiceName(),
+ memberSuspendedEvent.getClusterId());
+ }
+ }
+ });
+
+ addEventListener(new MemberTerminatedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ MemberTerminatedEvent memberTerminatedEvent = (MemberTerminatedEvent) event;
+
+ String serviceName = memberTerminatedEvent.getServiceName();
+ String clusterId = memberTerminatedEvent.getClusterId();
+ String memberId = memberTerminatedEvent.getMemberId();
+
+ try {
+ TopologyManager.acquireReadLockForCluster(serviceName, clusterId);
+ removeMember(serviceName, clusterId, memberId);
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLockForCluster(serviceName, clusterId);
+ }
+ }
+ });
+
+ addEventListener(new ClusterRemovedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ ClusterRemovedEvent clusterRemovedEvent = (ClusterRemovedEvent) event;
+ String serviceName = clusterRemovedEvent.getServiceName();
+ String clusterId = clusterRemovedEvent.getClusterId();
+
+ try {
+ TopologyManager.acquireReadLockForCluster(serviceName, clusterId);
+
+ Service service = TopologyManager.getTopology().getService(serviceName);
+ if (service == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Service not found in topology: [service] %s", serviceName));
+ }
+ return;
+ }
+
+ Cluster cluster = service.getCluster(clusterId);
+ removeCluster(cluster);
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLockForCluster(serviceName, clusterId);
+ }
+ }
+ });
+
+ addEventListener(new ServiceRemovedEventListener() {
+ @Override
+ protected void onEvent(Event event) {
+
+ ServiceRemovedEvent serviceRemovedEvent = (ServiceRemovedEvent) event;
+ String serviceName = serviceRemovedEvent.getServiceName();
+
+ try {
+ TopologyManager.acquireReadLockForService(serviceName);
+
+ Service service = TopologyManager.getTopology().getService(serviceName);
+ if (service == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Service not found in topology: [service] %s",
+ serviceName));
+ }
+ return;
+ }
+ for(Cluster cluster : service.getClusters()) {
+ removeCluster(cluster);
+ }
+ } catch (Exception e) {
+ log.error("Error processing event", e);
+ } finally {
+ TopologyManager.releaseReadLockForService(serviceName);
+ }
+ }
+ });
+ }
+
+ /**
+ * Remove cluster from topology provider
+ * @param cluster
+ */
+ private void removeCluster(Cluster cluster) {
+ for(Member member : cluster.getMembers()) {
+ removeMember(member.getServiceName(), member.getClusterId(), member.getMemberId());
+ }
+ }
+
+ /**
+ * Add member to topology provider
+ * @param serviceName
+ * @param clusterId
+ * @param memberId
+ */
+ private void addMember(String serviceName, String clusterId, String memberId) {
+ Service service = TopologyManager.getTopology().getService(serviceName);
+ if (service == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Service not found in topology: [service] %s",
+ serviceName));
+ }
+ return;
+ }
+ Cluster cluster = service.getCluster(clusterId);
+ if (cluster == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Cluster not found in topology: [service] %s [cluster] %s",
+ serviceName, clusterId));
+ }
+ return;
+ }
+ validateHostNames(cluster);
+
+ // Add cluster if not exists
+ if(!topologyProvider.clusterExistsByClusterId(cluster.getClusterId())) {
+ topologyProvider.addCluster(transformCluster(cluster));
+ }
+
+ Member member = cluster.getMember(memberId);
+ if (member == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Member not found in topology: [service] %s [cluster] %s [member] %s",
+ serviceName, clusterId,
+ memberId));
+ }
+ return;
+ }
+ topologyProvider.addMember(transformMember(member));
+ }
+
+ /**
+ * Remove member from topology provider
+ * @param serviceName
+ * @param clusterId
+ * @param memberId
+ */
+ private void removeMember(String serviceName, String clusterId, String memberId) {
+ Service service = TopologyManager.getTopology().getService(serviceName);
+ if (service == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Service not found in topology: [service] %s",
+ serviceName));
+ }
+ return;
+ }
+
+ Cluster cluster = service.getCluster(clusterId);
+ if (cluster == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Cluster not found in topology: [service] %s [cluster] %s",
+ serviceName, clusterId));
+ }
+ return;
+ }
+ validateHostNames(cluster);
+
+ Member member = cluster.getMember(memberId);
+ if (member == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Member not found in topology: [service] %s [cluster] %s [member] %s",
+ serviceName, clusterId,
+ memberId));
+ }
+ return;
+ }
+
+ if (member != null) {
+ topologyProvider.removeMember(cluster.getClusterId(), member.getMemberId());
+ }
+ }
+
+ private void validateHostNames(Cluster cluster) {
+ if((cluster.getHostNames() == null) || (cluster.getHostNames().size() == 0)) {
+ throw new RuntimeException(String.format("Host names not found in cluster: " +
+ "[cluster] %s", cluster.getClusterId()));
+ }
+ }
+
+ private org.apache.stratos.load.balancer.common.domain.Cluster transformCluster(Cluster messagingCluster) {
+ org.apache.stratos.load.balancer.common.domain.Cluster cluster =
+ new org.apache.stratos.load.balancer.common.domain.Cluster(messagingCluster.getServiceName(),
+ messagingCluster.getClusterId());
+ cluster.setTenantRange(messagingCluster.getTenantRange());
+ if(messagingCluster.getHostNames() != null) {
+ for (String hostName : messagingCluster.getHostNames()) {
+ cluster.addHostName(hostName);
+ }
+ }
+ return cluster;
+ }
+
+ private org.apache.stratos.load.balancer.common.domain.Member transformMember(Member messagingMember) {
+ boolean useMemberPublicIP = Boolean.getBoolean("load.balancer.member.public.ip");
+ String hostName = (useMemberPublicIP) ? messagingMember.getDefaultPublicIP() :
+ messagingMember.getDefaultPrivateIP();
+ org.apache.stratos.load.balancer.common.domain.Member member =
+ new org.apache.stratos.load.balancer.common.domain.Member(messagingMember.getServiceName(),
+ messagingMember.getClusterId(), messagingMember.getMemberId(),
+ hostName);
+ return member;
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/LoadBalancerStatisticsReader.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/LoadBalancerStatisticsReader.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/LoadBalancerStatisticsReader.java
index 79386bd..c0016fe 100644
--- a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/LoadBalancerStatisticsReader.java
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/LoadBalancerStatisticsReader.java
@@ -27,11 +27,18 @@ import org.apache.stratos.messaging.domain.topology.Cluster;
public interface LoadBalancerStatisticsReader {
/**
- * Get in-flight request count of a sliding window configured e.g. Requests in flight of last minute.
+ * Returns in-flight request count of sliding window configured.
* @param clusterId
*/
int getInFlightRequestCount(String clusterId);
- int getServedRequestCount(String clusterId);
+
int getActiveInstancesCount(Cluster cluster);
+
+ /**
+ * Returns the number of requests served since the last time this method was called.
+ * @param clusterId
+ * @return
+ */
+ int getServedRequestCount(String clusterId);
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/topology/TopologyProvider.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/topology/TopologyProvider.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/topology/TopologyProvider.java
new file mode 100644
index 0000000..0eb77f8
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/topology/TopologyProvider.java
@@ -0,0 +1,175 @@
+/*
+ * 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.common.topology;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.load.balancer.common.domain.Cluster;
+import org.apache.stratos.load.balancer.common.domain.Member;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Load balancer topology provider.
+ */
+public class TopologyProvider {
+
+ private static final Log log = LogFactory.getLog(TopologyProvider.class);
+
+ private Map<String, Cluster> clusterIdToClusterMap;
+ private Map<String, Cluster> hostNameToClusterMap;
+ private Map<String, Map<Integer, Cluster>> hostNameToTenantIdToClusterMap;
+
+ public TopologyProvider() {
+ this.clusterIdToClusterMap = new ConcurrentHashMap<String, Cluster>();
+ this.hostNameToClusterMap = new ConcurrentHashMap<String, Cluster>();
+ this.hostNameToTenantIdToClusterMap = new ConcurrentHashMap<String, Map<Integer, Cluster>>();
+ }
+
+ public void addCluster(Cluster cluster) {
+ if(cluster != null) {
+ clusterIdToClusterMap.put(cluster.getClusterId(), cluster);
+
+ for(String hostName : cluster.getHostNames()) {
+ hostNameToClusterMap.put(hostName, cluster);
+ }
+
+ if((cluster.getHostNames() != null) && (cluster.getHostNames().size() > 0)) {
+ log.info(String.format("Cluster added: [cluster] %s [hostnames] %s", cluster.getClusterId(),
+ cluster.getHostNames()));
+ }
+ }
+ }
+
+ public void removeCluster(String clusterId) {
+ Cluster cluster = getClusterByClusterId(clusterId);
+ if(cluster == null) {
+ log.warn(String.format("Could not remove cluster, cluster not found: [cluster] %s", clusterId));
+ return;
+ }
+
+ for(String hostName : cluster.getHostNames()) {
+ hostNameToClusterMap.remove(hostName);
+ }
+ clusterIdToClusterMap.remove(cluster.getClusterId());
+
+ if((cluster.getHostNames() != null) && (cluster.getHostNames().size() > 0)) {
+ log.info(String.format("Cluster removed: [cluster] %s [hostnames] %s", cluster.getClusterId(),
+ cluster.getHostNames()));
+ }
+ }
+
+ public boolean clusterExistsByClusterId(String clusterId) {
+ return (getClusterByClusterId(clusterId) != null);
+ }
+
+ public boolean clusterExistsByHostName(String hostName) {
+ return (hostNameToClusterMap.containsKey(hostName) || hostNameToTenantIdToClusterMap.containsKey(hostName));
+ }
+
+ public Cluster getClusterByClusterId(String clusterId) {
+ return clusterIdToClusterMap.get(clusterId);
+ }
+
+ public void addCluster(Cluster cluster, int tenantId) {
+ if(cluster != null) {
+ boolean subscribed = false;
+ for(String hostName : cluster.getHostNames()) {
+ Map<Integer, Cluster> tenantIdToClusterMap = hostNameToTenantIdToClusterMap.get(hostName);
+ if(tenantIdToClusterMap == null) {
+ tenantIdToClusterMap = new ConcurrentHashMap<Integer, Cluster>();
+ hostNameToTenantIdToClusterMap.put(hostName, tenantIdToClusterMap);
+ }
+ tenantIdToClusterMap.put(tenantId, cluster);
+ subscribed = true;
+ }
+ if(subscribed) {
+ log.info(String.format("Tenant subscribed to cluster: [tenant] %d [cluster] %s [hostnames] %s",
+ tenantId, cluster.getClusterId(), cluster.getHostNames()));
+ }
+ }
+ }
+
+ public void removeCluster(String clusterId, int tenantId) {
+ Cluster cluster = getClusterByClusterId(clusterId);
+ if(cluster == null) {
+ log.warn(String.format("Could not remove cluster, cluster not found: [cluster] %s", clusterId));
+ }
+
+ for(String hostName : cluster.getHostNames()) {
+ Map<Integer, Cluster> tenantIdToClusterMap = hostNameToTenantIdToClusterMap.get(hostName);
+ if (tenantIdToClusterMap != null) {
+ Cluster cluster_ = tenantIdToClusterMap.get(tenantId);
+ if (cluster_ != null) {
+ tenantIdToClusterMap.remove(tenantId);
+ log.info(String.format("Tenant un-subscribed from cluster: [tenant] %d [cluster] %s [hostnames] %s",
+ tenantId, cluster.getClusterId(), cluster.getHostNames()));
+ }
+ }
+ }
+ }
+
+ public Cluster getClusterByHostName(String hostName) {
+ return hostNameToClusterMap.get(hostName);
+ }
+
+ public Cluster getClusterByHostName(String hostName, int tenantId) {
+ Map<Integer, Cluster> tenantIdToClusterMap = hostNameToTenantIdToClusterMap.get(hostName);
+ if(tenantIdToClusterMap != null) {
+ return tenantIdToClusterMap.get(tenantId);
+ }
+ return null;
+ }
+
+ public void addMember(Member member) {
+ Cluster cluster = getClusterByClusterId(member.getClusterId());
+ if(cluster == null) {
+ log.warn(String.format("Could not add member, cluster not found: [cluster] %s",
+ member.getClusterId()));
+ return;
+ }
+
+ cluster.addMember(member);
+ log.info(String.format("Member added to cluster: [cluster] %s [member] %s",
+ member.getClusterId(), member.getHostName()));
+ }
+
+ public void removeMember(String clusterId, String memberId) {
+ Cluster cluster = getClusterByClusterId(clusterId);
+ if(cluster == null) {
+ log.warn(String.format("Could not remove member, cluster not found: [cluster] %s", clusterId));
+ return;
+ }
+
+ Member member = cluster.getMember(memberId);
+ if(member != null) {
+ cluster.removeMember(memberId);
+ log.info(String.format("Member removed from cluster: [cluster] %s [member] %s",
+ clusterId, member.getHostName()));
+
+ if(cluster.getMembers().size() == 0) {
+ log.info(String.format("No members found in cluster, removing cluster: " +
+ "[cluster] %s", cluster.getClusterId()));
+ removeCluster(cluster.getClusterId());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java
index d561eb6..1344c0b 100755
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java
@@ -20,7 +20,7 @@
package org.apache.stratos.load.balancer.algorithm;
import org.apache.stratos.load.balancer.context.AlgorithmContext;
-import org.apache.stratos.messaging.domain.topology.Member;
+import org.apache.stratos.load.balancer.common.domain.Member;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java
index f96e44c..af08f79 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java
@@ -22,7 +22,7 @@ package org.apache.stratos.load.balancer.algorithm;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.load.balancer.context.AlgorithmContext;
-import org.apache.stratos.messaging.domain.topology.Member;
+import org.apache.stratos.load.balancer.common.domain.Member;
import java.util.List;
import java.util.concurrent.locks.Lock;
@@ -67,7 +67,7 @@ public class RoundRobin implements LoadBalanceAlgorithm {
currentMemberIndex++;
}
index--;
- } while ((!current.isActive()) && index > 0);
+ } while (index > 0);
algorithmContext.setCurrentMemberIndex(currentMemberIndex);
} finally {
lock.unlock();
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/conf/LoadBalancerConfiguration.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/conf/LoadBalancerConfiguration.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/conf/LoadBalancerConfiguration.java
index 9e86a6e..658ce54 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/conf/LoadBalancerConfiguration.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/conf/LoadBalancerConfiguration.java
@@ -23,6 +23,10 @@ import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.common.constants.StratosConstants;
+import org.apache.stratos.load.balancer.common.domain.Cluster;
+import org.apache.stratos.load.balancer.common.domain.Member;
+import org.apache.stratos.load.balancer.common.domain.Port;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
import org.apache.stratos.load.balancer.conf.domain.Algorithm;
import org.apache.stratos.load.balancer.conf.domain.MemberIpType;
import org.apache.stratos.load.balancer.conf.domain.TenantIdentifier;
@@ -30,10 +34,7 @@ import org.apache.stratos.load.balancer.conf.structure.Node;
import org.apache.stratos.load.balancer.conf.structure.NodeBuilder;
import org.apache.stratos.load.balancer.conf.util.Constants;
import org.apache.stratos.load.balancer.context.LoadBalancerContext;
-import org.apache.stratos.load.balancer.context.LoadBalancerContextUtil;
import org.apache.stratos.load.balancer.exception.InvalidConfigurationException;
-import org.apache.stratos.messaging.domain.topology.*;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
import java.io.File;
import java.util.*;
@@ -66,6 +67,7 @@ public class LoadBalancerConfiguration {
private String networkPartitionId;
private boolean reWriteLocationHeader;
private boolean domainMappingEnabled;
+ private TopologyProvider topologyProvider;
/**
* Load balancer configuration is singleton.
@@ -100,8 +102,6 @@ public class LoadBalancerConfiguration {
instance = null;
// Clear load balancer context
LoadBalancerContext.getInstance().clear();
- // Clear topology
- TopologyManager.getTopology().clear();
}
}
@@ -269,6 +269,14 @@ public class LoadBalancerConfiguration {
this.domainMappingEnabled = domainMappingEnabled;
}
+ public void setTopologyProvider(TopologyProvider topologyProvider) {
+ this.topologyProvider = topologyProvider;
+ }
+
+ public TopologyProvider getTopologyProvider() {
+ return topologyProvider;
+ }
+
private static class LoadBalancerConfigurationReader {
public LoadBalancerConfiguration readFromFile() {
@@ -323,7 +331,7 @@ public class LoadBalancerConfiguration {
} else {
// Endpoint timeout is not found, set default value
configuration.setEndpointTimeout(Constants.DEFAULT_ENDPOINT_TIMEOUT);
- if(log.isWarnEnabled()) {
+ if (log.isWarnEnabled()) {
log.warn(String.format("Endpoint timeout not found, using default: %d", configuration.getEndpointTimeout()));
}
}
@@ -334,7 +342,7 @@ public class LoadBalancerConfiguration {
} else {
// Session timeout is not found, set default value
configuration.setSessionTimeout(Constants.DEFAULT_SESSION_TIMEOUT);
- if(log.isWarnEnabled()) {
+ if (log.isWarnEnabled()) {
log.warn(String.format("Session timeout not found, using default: %d", configuration.getSessionTimeout()));
}
}
@@ -343,8 +351,8 @@ public class LoadBalancerConfiguration {
validateRequiredPropertyInNode(Constants.CONF_PROPERTY_TOPOLOGY_EVENT_LISTENER, topologyEventListenerEnabled, Constants.CONF_ELEMENT_LOADBALANCER);
configuration.setTopologyEventListenerEnabled(Boolean.parseBoolean(topologyEventListenerEnabled));
- if(configuration.isTopologyEventListenerEnabled()) {
- String topologyMemberIpType = loadBalancerNode.getProperty(Constants.CONF_PROPERTY_TOPOLOGY_MEMBER_IP_TYPE);
+ if (configuration.isTopologyEventListenerEnabled()) {
+ String topologyMemberIpType = loadBalancerNode.getProperty(Constants.CONF_PROPERTY_TOPOLOGY_MEMBER_IP_TYPE);
validateRequiredPropertyInNode(Constants.CONF_PROPERTY_TOPOLOGY_MEMBER_IP_TYPE, topologyMemberIpType, Constants.CONF_ELEMENT_LOADBALANCER);
configuration.setTopologyMemberIpType(transformMemberIpType(topologyMemberIpType));
}
@@ -366,7 +374,7 @@ public class LoadBalancerConfiguration {
}
String clusterFilter = loadBalancerNode.getProperty(Constants.CONF_PROPERTY_TOPOLOGY_CLUSTER_FILTER);
if (StringUtils.isNotBlank(clusterFilter)) {
- log.info("Cluster filter::: " +clusterFilter);
+ log.info("Cluster filter::: " + clusterFilter);
configuration.setTopologyClusterFilter(clusterFilter);
}
String memberFilter = loadBalancerNode.getProperty(Constants.CONF_PROPERTY_TOPOLOGY_MEMBER_FILTER);
@@ -410,11 +418,11 @@ public class LoadBalancerConfiguration {
throw new InvalidConfigurationException(String.format("Invalid tenant identifier regular expression: %s", tenantIdentifierRegex), e);
}
List<String> regexList = new ArrayList<String>();
- if(tenantIdentifierRegex.contains(StratosConstants.FILTER_VALUE_SEPARATOR)) {
+ if (tenantIdentifierRegex.contains(StratosConstants.FILTER_VALUE_SEPARATOR)) {
String[] regexArray;
regexArray = tenantIdentifierRegex.split(StratosConstants.FILTER_VALUE_SEPARATOR);
- for(String regex: regexArray) {
- regexList.add(regex);
+ for (String regex : regexArray) {
+ regexList.add(regex);
}
} else {
regexList.add(tenantIdentifierRegex);
@@ -433,12 +441,12 @@ public class LoadBalancerConfiguration {
}
String rewriteLocationHeader = loadBalancerNode.getProperty(Constants.CONF_PROPERTY_REWRITE_LOCATION_HEADER);
- if(StringUtils.isNotEmpty(rewriteLocationHeader)) {
+ if (StringUtils.isNotEmpty(rewriteLocationHeader)) {
configuration.setRewriteLocationHeader(Boolean.parseBoolean(topologyEventListenerEnabled));
}
String mapDomainNames = loadBalancerNode.getProperty(Constants.CONF_PROPERTY_MAP_DOMAIN_NAMES);
- if(StringUtils.isNotEmpty(mapDomainNames)) {
+ if (StringUtils.isNotEmpty(mapDomainNames)) {
configuration.setDomainMappingEnabled(Boolean.parseBoolean(mapDomainNames));
}
@@ -446,24 +454,26 @@ public class LoadBalancerConfiguration {
Node servicesNode = loadBalancerNode.findChildNodeByName(Constants.CONF_ELEMENT_SERVICES);
validateRequiredNode(servicesNode, Constants.CONF_ELEMENT_SERVICES);
+ TopologyProvider topologyProvider = new TopologyProvider();
for (Node serviceNode : servicesNode.getChildNodes()) {
- ServiceType serviceType = ServiceType.SingleTenant;
+ boolean isMultiTenant = false;
String multiTenant = serviceNode.getProperty(Constants.CONF_PROPERTY_MULTI_TENANT);
if (StringUtils.isNotBlank(multiTenant) && (Boolean.parseBoolean(multiTenant))) {
- serviceType = ServiceType.MultiTenant;
+ isMultiTenant = true;
}
- Service service = new Service(serviceNode.getName(), serviceType);
+
+ String serviceName = serviceNode.getName();
Node clustersNode = serviceNode.findChildNodeByName(Constants.CONF_ELEMENT_CLUSTERS);
for (Node clusterNode : clustersNode.getChildNodes()) {
String clusterId = clusterNode.getName();
- Cluster cluster = new Cluster(service.getServiceName(), clusterId, null, null, null);
+ Cluster cluster = new Cluster(serviceName, clusterId);
String tenantRange = clusterNode.getProperty(Constants.CONF_PROPERTY_TENANT_RANGE);
if (StringUtils.isNotBlank(tenantRange)) {
- if (service.getServiceType() != ServiceType.MultiTenant) {
+ if (!isMultiTenant) {
throw new InvalidConfigurationException(String.format("%s property is not valid for non multi-tenant service cluster: [service] %s [cluster] %s",
- Constants.CONF_PROPERTY_TENANT_RANGE, service.getServiceName(), cluster.getClusterId()));
+ Constants.CONF_PROPERTY_TENANT_RANGE, serviceName, cluster.getClusterId()));
}
cluster.setTenantRange(tenantRange);
}
@@ -488,14 +498,10 @@ public class LoadBalancerConfiguration {
String memberId = memberNode.getName();
// we are making it as 1 because we are not using this for static loadbalancer configuration
long initTime = -1;
- Member member = new Member(cluster.getServiceName(), cluster.getClusterId(), memberId,
- memberId, Constants.STATIC_NETWORK_PARTITION, Constants.STATIC_PARTITION, initTime);
String ip = memberNode.getProperty(Constants.CONF_PROPERTY_IP);
validateRequiredPropertyInNode(Constants.CONF_PROPERTY_IP, ip, String.format("member %s", memberId));
- List<String> memberPrivateIPs = new ArrayList<String>();
- memberPrivateIPs.add(ip);
- member.setMemberPrivateIPs(memberPrivateIPs);
- member.setDefaultPrivateIP(ip);
+ Member member = new Member(cluster.getServiceName(), cluster.getClusterId(), memberId, ip);
+
Node portsNode = memberNode.findChildNodeByName(Constants.CONF_ELEMENT_PORTS);
validateRequiredNode(portsNode, Constants.CONF_ELEMENT_PORTS, String.format("member %s", memberId));
@@ -509,29 +515,13 @@ public class LoadBalancerConfiguration {
Port port = new Port(portNode.getName(), Integer.valueOf(value), Integer.valueOf(proxy));
member.addPort(port);
}
- member.setStatus(MemberStatus.Active);
cluster.addMember(member);
}
// Add cluster to service
- service.addCluster(cluster);
-
- // Add service to topology manager if not exists
- try {
- // TODO - fix properly!
- // this lock is not needed since, this Topology is not shared. This is
- // used by LB only
- //TopologyManager.acquireWriteLock();
- if (!TopologyManager.getTopology().serviceExists(service.getServiceName())) {
- TopologyManager.getTopology().addService(service);
- }
- } finally {
- //TopologyManager.releaseWriteLock();
- }
-
- // Add cluster to load balancer context
- LoadBalancerContextUtil.addClusterAgainstHostNames(cluster);
+ topologyProvider.addCluster(cluster);
}
}
+ configuration.setTopologyProvider(topologyProvider);
}
return configuration;
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContext.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContext.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContext.java
index 895d0d9..0eec59e 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContext.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContext.java
@@ -32,41 +32,25 @@ public class LoadBalancerContext {
private static volatile LoadBalancerContext instance;
- // Following map is updated by the service component.
- // Map<TenantId, SynapseEnvironmentService>
- private TenantIdSynapseEnvironmentServiceMap tenantIdSynapseEnvironmentServiceMap;
-
- // Following maps are updated on demand by the request delegator.
- // Map<ServiceName, ServiceContext>
- private ServiceNameServiceContextMap serviceNameServiceContextMap;
// Map<ClusterId, ClusterContext>
private ClusterIdClusterContextMap clusterIdClusterContextMap;
// Following maps are updated by load balancer topology & tenant receivers.
// Map<ClusterId, Cluster>
// Keep track of all clusters
- private ClusterIdClusterMap clusterIdClusterMap;
// Map<Host/Domain-Name, Cluster>
// Keep tack of all clusters
- private HostNameClusterMap hostNameClusterMap;
+
// Map<Host/Domain-Name, AppContext>
private HostNameAppContextMap hostNameAppContextMap;
// Map<HostName, Map<TenantId, Cluster>>
- // Keep track of multi-tenant service clusters
- private MultiTenantClusterMap multiTenantClusterMap;
// Map<MemberIp, Hostname>
// Keep track of cluster hostnames of of all members against their ip addresses
private MemberIpHostnameMap memberIpHostnameMap;
- private boolean clustered;
private LoadBalancerContext() {
- tenantIdSynapseEnvironmentServiceMap = new TenantIdSynapseEnvironmentServiceMap();
- serviceNameServiceContextMap = new ServiceNameServiceContextMap();
clusterIdClusterContextMap = new ClusterIdClusterContextMap();
- clusterIdClusterMap = new ClusterIdClusterMap();
- hostNameClusterMap = new HostNameClusterMap();
hostNameAppContextMap = new HostNameAppContextMap();
- multiTenantClusterMap = new MultiTenantClusterMap();
memberIpHostnameMap = new MemberIpHostnameMap();
}
@@ -82,49 +66,20 @@ public class LoadBalancerContext {
}
public void clear() {
- tenantIdSynapseEnvironmentServiceMap.clear();
- serviceNameServiceContextMap.clear();
clusterIdClusterContextMap.clear();
- multiTenantClusterMap.clear();
- }
-
- public TenantIdSynapseEnvironmentServiceMap getTenantIdSynapseEnvironmentServiceMap() {
- return tenantIdSynapseEnvironmentServiceMap;
- }
-
- public ServiceNameServiceContextMap getServiceNameServiceContextMap() {
- return serviceNameServiceContextMap;
+ hostNameAppContextMap.clear();
+ memberIpHostnameMap.clear();
}
public ClusterIdClusterContextMap getClusterIdClusterContextMap() {
return clusterIdClusterContextMap;
}
- public ClusterIdClusterMap getClusterIdClusterMap() {
- return clusterIdClusterMap;
- }
-
- public HostNameClusterMap getHostNameClusterMap() {
- return hostNameClusterMap;
- }
-
public HostNameAppContextMap getHostNameContextPathMap() {
return hostNameAppContextMap;
}
- public MultiTenantClusterMap getMultiTenantClusterMap() {
- return multiTenantClusterMap;
- }
-
public MemberIpHostnameMap getMemberIpHostnameMap() {
return memberIpHostnameMap;
}
-
- public boolean isClustered() {
- return clustered;
- }
-
- public void setClustered(boolean clustered) {
- this.clustered = clustered;
- }
}
[2/3] stratos git commit: Introducing load balancer topology provider
model to be able to provide the topology required for load balancing in a
generic manner
Posted by im...@apache.org.
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContextUtil.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContextUtil.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContextUtil.java
deleted file mode 100644
index 3fec1dd..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/LoadBalancerContextUtil.java
+++ /dev/null
@@ -1,475 +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.context;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.tenant.Subscription;
-import org.apache.stratos.messaging.domain.tenant.SubscriptionDomain;
-import org.apache.stratos.messaging.domain.tenant.Tenant;
-import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.Service;
-import org.apache.stratos.messaging.message.receiver.tenant.TenantManager;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * Load balancer context utility class.
- */
-public class LoadBalancerContextUtil {
- private static final Log log = LogFactory.getLog(LoadBalancerContextUtil.class);
-
- /**
- * Add cluster against its host names.
- *
- * @param cluster
- */
- public static void addClusterAgainstHostNames(Cluster cluster) {
- if (cluster == null)
- return;
-
- Service service = TopologyManager.getTopology().getService(cluster.getServiceName());
- if (service == null) {
- throw new RuntimeException(String.format("Service not found: [cluster] %s", cluster.getClusterId()));
- }
-
- // Add cluster to ClusterIdClusterMap
- LoadBalancerContext.getInstance().getClusterIdClusterMap().addCluster(cluster);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster added to cluster-id -> cluster map: [cluster] %s ", cluster.getClusterId()));
- }
-
- // Add cluster to HostNameClusterMap
- for (String hostName : cluster.getHostNames()) {
- addClusterToHostNameClusterMap(hostName, cluster);
-
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster added to host/domain name -> cluster map: [hostName] %s [cluster] %s",
- hostName, cluster.getClusterId()));
- }
- }
- }
-
- /**
- * Remove cluster mapped against its host names.
- *
- * @param clusterId
- */
- public static void removeClusterAgainstHostNames(String clusterId) {
- Cluster cluster = LoadBalancerContext.getInstance().getClusterIdClusterMap().getCluster(clusterId);
- if (cluster == null) {
- return;
- }
-
- Service service = TopologyManager.getTopology().getService(cluster.getServiceName());
- if (service == null) {
- throw new RuntimeException(String.format("Service not found: [cluster] %s", cluster.getClusterId()));
- }
-
- // Remove cluster from HostNameClusterMap
- for (String hostName : cluster.getHostNames()) {
- removeClusterFromHostNameClusterMap(hostName, cluster);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster removed from host/domain name -> clusters map: [host-name] %s [cluster] %s",
- hostName, cluster.getClusterId()));
- }
- }
-
- // Remove cluster from ClusterIdClusterMap
- LoadBalancerContext.getInstance().getClusterIdClusterMap().removeCluster(clusterId);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster removed from cluster-id -> cluster map: [cluster] %s ", cluster.getClusterId()));
- }
- }
-
- /**
- * Add clusters against host names, tenant id for the given service, cluster ids.
- *
- * @param serviceName
- * @param tenantId
- * @param clusterIds
- */
- public static void addClustersAgainstHostNamesAndTenantIds(String serviceName, int tenantId, Set<String> clusterIds) {
- try {
- TopologyManager.acquireReadLock();
-
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Service not found in topology: [service] %s", serviceName));
- }
- return;
- }
- Cluster cluster;
- for (String clusterId : clusterIds) {
- cluster = service.getCluster(clusterId);
- if (cluster != null) {
- // Add cluster against host names and tenant id
- addClusterAgainstHostNamesAndTenantId(serviceName, tenantId, cluster);
- } else {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Cluster not found in service: [service] %s [cluster] %s", serviceName, clusterId));
- }
- }
- }
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- /**
- * Remove clusters mapped against host names and tenant id.
- *
- * @param serviceName
- * @param tenantId
- */
- public static void removeClustersAgainstHostNamesAndTenantIds(String serviceName, int tenantId, Set<String> clusterIds) {
- try {
- TopologyManager.acquireReadLock();
-
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Service not found in topology: [service] %s", serviceName));
- }
- return;
- }
- Cluster cluster;
- for (String clusterId : clusterIds) {
- cluster = service.getCluster(clusterId);
- if (cluster != null) {
- // Remove cluster mapped against host names and tenant id
- removeClusterAgainstHostNamesAndTenantId(serviceName, tenantId, cluster);
- } else {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Cluster not found in service: [service] %s [cluster] %s", serviceName, clusterId));
- }
- }
- }
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- /**
- * Add clusters against domain name for the given service, cluster ids.
- *
- * @param serviceName
- * @param clusterId
- * @param domainName
- */
- public static void addClusterAgainstDomain(String serviceName, String clusterId, String domainName) {
- try {
- TopologyManager.acquireReadLock();
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Service not found in topology: [service] %s", serviceName));
- }
- return;
- }
- Cluster cluster = service.getCluster(clusterId);
- if (cluster != null) {
- addClusterAgainstDomain(serviceName, cluster, domainName);
- } else {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Cluster not found in service: [service] %s [cluster] %s", serviceName, clusterId));
- }
- }
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- /**
- * Remove clusters mapped against domain name for the given service, cluster ids.
- *
- * @param serviceName
- * @param clusterId
- * @param domainName
- */
- public static void removeClusterAgainstDomain(String serviceName, String clusterId, String domainName) {
- try {
- TopologyManager.acquireReadLock();
-
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Service not found in topology: [service] %s", serviceName));
- }
- return;
- }
- Cluster cluster = service.getCluster(clusterId);
- if (cluster != null) {
- // Remove clusters mapped against domain names
- removeClusterAgainstDomain(cluster, domainName);
- } else {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Cluster not found in service: [service] %s [cluster] %s", serviceName, clusterId));
- }
- }
- } finally {
- TopologyManager.releaseReadLock();
- }
- }
-
- /**
- * Find cluster from service name, tenant id.
- * Acquire a topology manager read lock appropriately.
- *
- * @param serviceName
- * @param tenantId
- * @return
- */
- @SuppressWarnings("unused")
- private static Cluster findCluster(String serviceName, int tenantId) {
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- throw new RuntimeException(String.format("Service not found: %s", serviceName));
- }
- for (Cluster cluster : service.getClusters()) {
- if (cluster.tenantIdInRange(tenantId)) {
- return cluster;
- }
- }
- return null;
- }
-
- /**
- * Add clusters against host names and tenant id to load balancer context.
- *
- * @param serviceName
- * @param tenantId
- * @param cluster
- */
- private static void addClusterAgainstHostNamesAndTenantId(String serviceName, int tenantId, Cluster cluster) {
- // Add clusters against host names
- if (log.isDebugEnabled()) {
- log.debug(String.format("Adding cluster to multi-tenant cluster map against host names: [service] %s " +
- "[tenant-id] %d [cluster] %s", serviceName, tenantId, cluster.getClusterId()));
- }
- for (String hostName : cluster.getHostNames()) {
- addClusterToMultiTenantClusterMap(hostName, tenantId, cluster);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster added to multi-tenant cluster map: [host-name] %s [tenant-id] %d [cluster] %s",
- hostName, tenantId, cluster.getClusterId()));
- }
- }
- }
-
- /**
- * Remove clusters mapped against host names and tenant id from load balancer context.
- *
- * @param serviceName
- * @param tenantId
- * @param cluster
- */
- private static void removeClusterAgainstHostNamesAndTenantId(String serviceName, int tenantId, Cluster cluster) {
- // Remove clusters mapped against host names
- if (log.isDebugEnabled()) {
- log.debug(String.format("Removing cluster from multi-tenant cluster map against host names: [service] %s " +
- "[tenant-id] %d [cluster] %s", serviceName, tenantId, cluster.getClusterId()));
- }
- for (String hostName : cluster.getHostNames()) {
- LoadBalancerContext.getInstance().getMultiTenantClusterMap().removeCluster(hostName, tenantId);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster removed from multi-tenant clusters map: [host-name] %s [tenant-id] %d [cluster] %s",
- hostName, tenantId, cluster.getClusterId()));
- }
- }
- }
-
-
- /**
- * Add clusters against domains to load balancer context.
- *
- * @param serviceName
- * @param cluster
- * @param domainName
- */
- private static void addClusterAgainstDomain(String serviceName, Cluster cluster, String domainName) {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Adding cluster to host/domain name -> cluster map against domain: [service] %s " +
- "[domain-name] %s [cluster] %s", serviceName, domainName, cluster.getClusterId()));
- }
- if (StringUtils.isNotBlank(domainName)) {
- addClusterToHostNameClusterMap(domainName, cluster);
-
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster added to host/domain name -> cluster map: [domain-name] %s [cluster] %s",
- domainName, cluster.getClusterId()));
- }
- }
- }
-
- /**
- * Remove clusters mapped against all subscription domain names for the given service, tenant, cluster ids.
- *
- * @param serviceName
- * @param tenantId
- * @param clusterIds
- */
- public static void removeClustersAgainstAllDomains(String serviceName, int tenantId, Set<String> clusterIds) {
- try {
- TenantManager.acquireReadLock();
- TopologyManager.acquireReadLock();
-
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Service not found in topology: [service] %s", serviceName));
- }
- return;
- }
- for (String clusterId : clusterIds) {
- Cluster cluster = service.getCluster(clusterId);
- Tenant tenant = TenantManager.getInstance().getTenant(tenantId);
- if (tenant != null) {
- for (Subscription subscription : tenant.getSubscriptions()) {
- if (subscription.getServiceName().equals(serviceName)) {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Removing cluster from host/domain name -> cluster map: [service] %s " +
- "[tenant-id] %d [domains] %s", serviceName, tenantId, subscription.getSubscriptionDomains()));
- }
- for (SubscriptionDomain subscriptionDomain : subscription.getSubscriptionDomains()) {
- removeClusterAgainstDomain(cluster, subscriptionDomain.getDomainName());
- }
- } else {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Tenant not subscribed to service: %s", serviceName));
- }
- }
- }
- }
- }
- } finally {
- TopologyManager.releaseReadLock();
- TenantManager.releaseReadLock();
- }
- }
-
- private static void removeClusterAgainstDomain(Cluster cluster, String domainName) {
- removeClusterFromHostNameClusterMap(domainName, cluster);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Cluster removed from host/domain name -> cluster map: [domain-name] %s [cluster] %s",
- domainName, cluster.getClusterId()));
- }
- }
-
- /**
- * Add cluster to host/domain name cluster map.
- *
- * @param hostName
- * @param cluster
- */
- private static void addClusterToHostNameClusterMap(String hostName, Cluster cluster) {
- if (!LoadBalancerContext.getInstance().getHostNameClusterMap().containsCluster((hostName))) {
- LoadBalancerContext.getInstance().getHostNameClusterMap().addCluster(hostName, cluster);
- }
- }
-
- /**
- * Remove cluseter from host/domain names cluster map.
- *
- * @param hostName
- * @param cluster
- */
- private static void removeClusterFromHostNameClusterMap(String hostName, Cluster cluster) {
- if (LoadBalancerContext.getInstance().getHostNameClusterMap().containsCluster(hostName)) {
- LoadBalancerContext.getInstance().getHostNameClusterMap().removeCluster(hostName);
- }
- }
-
- /**
- * Add cluster to multi-tenant cluster map.
- *
- * @param hostName
- * @param tenantId
- * @param cluster
- */
- private static void addClusterToMultiTenantClusterMap(String hostName, int tenantId, Cluster cluster) {
- // Add hostName, tenantId, cluster to multi-tenant map
- Map<Integer, Cluster> clusterMap = LoadBalancerContext.getInstance().getMultiTenantClusterMap().getClusters(hostName);
- if (clusterMap == null) {
- clusterMap = new HashMap<Integer, Cluster>();
- clusterMap.put(tenantId, cluster);
- LoadBalancerContext.getInstance().getMultiTenantClusterMap().addClusters(hostName, clusterMap);
- } else {
- clusterMap.put(tenantId, cluster);
- }
- }
-
- public static void addContextPathAgainstDomain(String domainName, String appContext) {
- LoadBalancerContext.getInstance().getHostNameContextPathMap().addContextPath(domainName, appContext);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Context path added against domain name: [domain-name] %s [context-path] %s",
- domainName, appContext));
- }
- }
-
- public static void removeContextPathAgainstDomain(String domainName) {
- LoadBalancerContext.getInstance().getHostNameContextPathMap().removeContextPath(domainName);
- if (log.isDebugEnabled()) {
- log.debug(String.format("Context path removed against domain name: [domain-name] %s",
- domainName));
- }
- }
-
- public static void removeAppContextAgainstAllDomains(String serviceName, int tenantId) {
- try {
- TenantManager.acquireReadLock();
- TopologyManager.acquireReadLock();
-
- Service service = TopologyManager.getTopology().getService(serviceName);
- if (service == null) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Service not found in topology: [service] %s", serviceName));
- }
- return;
- }
-
- Tenant tenant = TenantManager.getInstance().getTenant(tenantId);
- if (tenant != null) {
- for (Subscription subscription : tenant.getSubscriptions()) {
- if (subscription.getServiceName().equals(serviceName)) {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Removing appContext against domain name: [service] %s " +
- "[tenant-id] %d [domains] %s", serviceName, tenantId, subscription.getSubscriptionDomains()));
- }
- for (SubscriptionDomain subscriptionDomain : subscription.getSubscriptionDomains()) {
- removeContextPathAgainstDomain(subscriptionDomain.getDomainName());
- }
- } else {
- if (log.isDebugEnabled()) {
- log.debug(String.format("Tenant not subscribed to service: %s", serviceName));
- }
- }
- }
- }
- } finally {
- TopologyManager.releaseReadLock();
- TenantManager.releaseReadLock();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/ClusterIdClusterMap.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/ClusterIdClusterMap.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/ClusterIdClusterMap.java
deleted file mode 100644
index 958c1aa..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/ClusterIdClusterMap.java
+++ /dev/null
@@ -1,57 +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.context.map;
-
-import org.apache.stratos.messaging.domain.topology.Cluster;
-
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * Cluster id, cluster map for accessing clusters using cluster id:
- * Map[ClusterId, Cluster]
- */
-public class ClusterIdClusterMap {
-
- private ConcurrentHashMap<String, Cluster> concurrentHashMap;
-
- public ClusterIdClusterMap() {
- concurrentHashMap = new ConcurrentHashMap<String, Cluster>();
- }
-
- public Cluster getCluster(String clusterId) {
- return concurrentHashMap.get(clusterId);
- }
-
- public boolean containsCluster(String clusterId) {
- return concurrentHashMap.containsKey(clusterId);
- }
-
- public void addCluster(Cluster cluster) {
- concurrentHashMap.put(cluster.getClusterId(), cluster);
- }
-
- public void removeCluster(String clusterId) {
- concurrentHashMap.remove(clusterId);
- }
-
- public void clear() {
- concurrentHashMap.clear();
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameAppContextMap.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameAppContextMap.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameAppContextMap.java
index 3e71093..993c41a 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameAppContextMap.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameAppContextMap.java
@@ -48,4 +48,8 @@ public class HostNameAppContextMap {
public boolean contains(String hostName) {
return concurrentHashMap.containsKey(hostName);
}
+
+ public void clear() {
+ concurrentHashMap.clear();
+ }
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameClusterMap.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameClusterMap.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameClusterMap.java
deleted file mode 100644
index c855dd3..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/HostNameClusterMap.java
+++ /dev/null
@@ -1,63 +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.context.map;
-
-import org.apache.stratos.messaging.domain.topology.Cluster;
-
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * Host/domain name cluster map for accessing clusters using host name:
- * Map[HostName, Cluster]
- */
-public class HostNameClusterMap {
-
- private ConcurrentHashMap<String, Cluster> concurrentHashMap;
-
- public HostNameClusterMap() {
- concurrentHashMap = new ConcurrentHashMap<String, Cluster>();
- }
-
- public Cluster getCluster(String hostName) {
- return concurrentHashMap.get(hostName);
- }
-
- public boolean containsCluster(String hostName) {
- return concurrentHashMap.containsKey(hostName);
- }
-
- public void addCluster(String hostName, Cluster cluster) {
- concurrentHashMap.put(hostName, cluster);
- }
-
- public void removeCluster(Cluster cluster) {
- for (String hostName : cluster.getHostNames()) {
- removeCluster(hostName);
- }
- }
-
- public void removeCluster(String hostName) {
- concurrentHashMap.remove(hostName);
- }
-
- public void clear() {
- concurrentHashMap.clear();
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/MultiTenantClusterMap.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/MultiTenantClusterMap.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/MultiTenantClusterMap.java
deleted file mode 100644
index d038fbc..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/MultiTenantClusterMap.java
+++ /dev/null
@@ -1,77 +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.context.map;
-
-import org.apache.stratos.messaging.domain.topology.Cluster;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * Multi-tenant cluster map for accessing clusters using host name and tenant id:
- * Map[HostName, Map[TenantId,Cluster]]
- */
-public class MultiTenantClusterMap {
-
- private ConcurrentHashMap<String, Map<Integer, Cluster>> concurrentHashMap;
-
- public MultiTenantClusterMap() {
- concurrentHashMap = new ConcurrentHashMap<String, Map<Integer, Cluster>>();
- }
-
- public Cluster getCluster(String hostName, int tenantId) {
- Map<Integer, Cluster> clusterMap = getClusters(hostName);
- if (clusterMap != null) {
- if(clusterMap.containsKey(tenantId)) {
- return clusterMap.get(tenantId);
- }
- }
- return null;
- }
-
- public Map<Integer, Cluster> getClusters(String hostName) {
- return concurrentHashMap.get(hostName);
- }
-
- public boolean containsClusters(String hostName) {
- return concurrentHashMap.containsKey(hostName);
- }
-
- public void addClusters(String hostname, Map<Integer, Cluster> clusters) {
- concurrentHashMap.put(hostname, clusters);
- }
-
- public void removeClusters(String hostName) {
- concurrentHashMap.remove(hostName);
- }
-
- public void clear() {
- concurrentHashMap.clear();
- }
-
- public void removeCluster(String hostName, int tenantId) {
- Map<Integer, Cluster> clusterMap = getClusters(hostName);
- if(clusterMap != null) {
- if(clusterMap.containsKey(tenantId)) {
- clusterMap.remove(tenantId);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/ServiceNameServiceContextMap.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/ServiceNameServiceContextMap.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/ServiceNameServiceContextMap.java
deleted file mode 100644
index 99d563b..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/ServiceNameServiceContextMap.java
+++ /dev/null
@@ -1,58 +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.context.map;
-
-import org.apache.stratos.load.balancer.context.ServiceContext;
-
-import java.util.Collection;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * Service name, service context map for accessing service contexts using service name:
- * Map[ServiceName, ServiceContext]
- */
-public class ServiceNameServiceContextMap {
-
- private ConcurrentHashMap<String, ServiceContext> concurrentHashMap;
-
- public ServiceNameServiceContextMap() {
- concurrentHashMap = new ConcurrentHashMap<String, ServiceContext>();
- }
-
- public Collection<ServiceContext> getServiceContexts() {
- return concurrentHashMap.values();
- }
-
- public ServiceContext getServiceContext(String serviceName) {
- return concurrentHashMap.get(serviceName);
- }
-
- public void addServiceContext(ServiceContext serviceContext) {
- concurrentHashMap.put(serviceContext.getServiceName(), serviceContext);
- }
-
- public void removeServiceContext(String serviceName) {
- concurrentHashMap.remove(serviceName);
- }
-
- public void clear() {
- concurrentHashMap.clear();
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/TenantIdSynapseEnvironmentServiceMap.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/TenantIdSynapseEnvironmentServiceMap.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/TenantIdSynapseEnvironmentServiceMap.java
deleted file mode 100644
index 7c59d07..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/context/map/TenantIdSynapseEnvironmentServiceMap.java
+++ /dev/null
@@ -1,63 +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.context.map;
-
-import org.wso2.carbon.mediation.initializer.services.SynapseEnvironmentService;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * Tenant id, synapse environment service map for accessing synapse environment services using
- * tenant id:
- * Map[TenantId, SynapseEnvironmentService]
- */
-public class TenantIdSynapseEnvironmentServiceMap {
-
- private ConcurrentHashMap<Integer, SynapseEnvironmentService> concurrentHashMap;
-
- public TenantIdSynapseEnvironmentServiceMap() {
- concurrentHashMap = new ConcurrentHashMap<Integer, SynapseEnvironmentService>();
- }
-
- public SynapseEnvironmentService getSynapseEnvironmentService(int tenantId) {
- return concurrentHashMap.get(tenantId);
- }
-
- public boolean containsSynapseEnvironmentService(int tenantId) {
- return concurrentHashMap.containsKey(tenantId);
- }
-
- public void addSynapseEnvironmentService(int tenantId, SynapseEnvironmentService synapseEnvironmentService) {
- concurrentHashMap.put(tenantId, synapseEnvironmentService);
- }
-
- public void removeSynapseEnvironmentService(int tenantId) {
- concurrentHashMap.remove(tenantId);
- }
-
- public Map<Integer, SynapseEnvironmentService> getTenantIdSynapseEnvironmentServiceMap() {
- return concurrentHashMap;
- }
-
- public void clear() {
- concurrentHashMap.clear();
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/RequestDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/RequestDelegator.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/RequestDelegator.java
index ac94654..99fcaa1 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/RequestDelegator.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/RequestDelegator.java
@@ -21,13 +21,13 @@ package org.apache.stratos.load.balancer.endpoint;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.load.balancer.context.AlgorithmContext;
import org.apache.stratos.load.balancer.algorithm.LoadBalanceAlgorithm;
-import org.apache.stratos.load.balancer.conf.LoadBalancerConfiguration;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
+import org.apache.stratos.load.balancer.common.domain.Cluster;
+import org.apache.stratos.load.balancer.common.domain.Member;
+import org.apache.stratos.load.balancer.context.AlgorithmContext;
import org.apache.stratos.load.balancer.context.ClusterContext;
import org.apache.stratos.load.balancer.context.LoadBalancerContext;
-import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.Member;
import java.util.ArrayList;
@@ -36,21 +36,30 @@ import java.util.ArrayList;
* according to the incoming request information.
*/
public class RequestDelegator {
+
private static final Log log = LogFactory.getLog(RequestDelegator.class);
private LoadBalanceAlgorithm algorithm;
+ private TopologyProvider topologyProvider;
- public RequestDelegator(LoadBalanceAlgorithm algorithm) {
+ public RequestDelegator(LoadBalanceAlgorithm algorithm, TopologyProvider topologyProvider) {
this.algorithm = algorithm;
+ this.topologyProvider = topologyProvider;
}
+ /**
+ * Find the next member in a cluster by applying a load balancing algorithm by the given host name.
+ * @param hostName host name of the cluster
+ * @param messageId synapse message id to be included in debugging logs
+ * @return
+ */
public Member findNextMemberFromHostName(String hostName, String messageId) {
if (hostName == null)
return null;
long startTime = System.currentTimeMillis();
- Cluster cluster = LoadBalancerContext.getInstance().getHostNameClusterMap().getCluster(hostName);
+ Cluster cluster = topologyProvider.getClusterByHostName(hostName);
if (cluster != null) {
if (log.isDebugEnabled()) {
log.debug(String.format("Cluster %s identified for request %s", cluster.getClusterId(), messageId));
@@ -59,7 +68,7 @@ public class RequestDelegator {
if (member != null) {
if (log.isDebugEnabled()) {
long endTime = System.currentTimeMillis();
- log.debug(String.format("Next member identified in %dms: [service] %s [cluster] %s [member] %s [request-id] %s",
+ log.debug(String.format("Next member identified in %dms: [service] %s [cluster] %s [member] %s [message-id] %s",
(endTime - startTime), member.getServiceName(), member.getClusterId(), member.getMemberId(), messageId));
}
}
@@ -73,11 +82,17 @@ public class RequestDelegator {
return null;
}
+ /**
+ * Find the next member in a cluster by applying a load balancing algorithm by the given host name and tenant id.
+ * @param hostName host name of the cluster
+ * @param tenantId tenant id of the incoming request
+ * @return
+ */
public Member findNextMemberFromTenantId(String hostName, int tenantId) {
long startTime = System.currentTimeMillis();
// Find cluster from host name and tenant id
- Cluster cluster = LoadBalancerContext.getInstance().getMultiTenantClusterMap().getCluster(hostName, tenantId);
+ Cluster cluster = topologyProvider.getClusterByHostName(hostName, tenantId);
if (cluster != null) {
Member member = findNextMemberInCluster(cluster);
if (member != null) {
@@ -98,13 +113,16 @@ public class RequestDelegator {
}
/**
+ * Find next member in the cluster by applying a load balancing algorithm.
+ *
* This operation should be synchronized in order to find a member
* correctly. This has no performance impact as per the load tests
* carried out.
*/
private synchronized Member findNextMemberInCluster(Cluster cluster) {
// Find algorithm context of the cluster
- ClusterContext clusterContext = LoadBalancerContext.getInstance().getClusterIdClusterContextMap().getClusterContext(cluster.getClusterId());
+ ClusterContext clusterContext = LoadBalancerContext.getInstance().
+ getClusterIdClusterContextMap().getClusterContext(cluster.getClusterId());
if (clusterContext == null) {
clusterContext = new ClusterContext(cluster.getServiceName(), cluster.getClusterId());
LoadBalancerContext.getInstance().getClusterIdClusterContextMap().addClusterContext(clusterContext);
@@ -119,7 +137,8 @@ public class RequestDelegator {
Member member = algorithm.getNextMember(algorithmContext);
if (member == null) {
if (log.isWarnEnabled()) {
- log.warn(String.format("Could not find a member in cluster: [service] %s [cluster] %s", cluster.getServiceName(), cluster.getClusterId()));
+ log.warn(String.format("Could not find a member in cluster: [service] %s [cluster] %s",
+ cluster.getServiceName(), cluster.getClusterId()));
}
}
return member;
@@ -129,10 +148,11 @@ public class RequestDelegator {
if (hostName == null)
return false;
- boolean valid = LoadBalancerContext.getInstance().getHostNameClusterMap().containsCluster(hostName);
- if ((!valid) && (LoadBalancerConfiguration.getInstance().isMultiTenancyEnabled())) {
- valid = LoadBalancerContext.getInstance().getMultiTenantClusterMap().containsClusters(hostName);
- }
+ boolean valid = topologyProvider.clusterExistsByHostName(hostName);
return valid;
}
+
+ public Cluster getCluster(String hostName) {
+ return topologyProvider.getClusterByHostName(hostName);
+ }
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/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 8e41807..8a35b12 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
@@ -23,9 +23,13 @@ import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.description.TransportInDescription;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.protocol.HTTP;
+import org.apache.stratos.load.balancer.algorithm.LoadBalanceAlgorithm;
import org.apache.stratos.load.balancer.algorithm.LoadBalanceAlgorithmFactory;
+import org.apache.stratos.load.balancer.common.domain.Cluster;
+import org.apache.stratos.load.balancer.common.domain.Member;
+import org.apache.stratos.load.balancer.common.domain.Port;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
import org.apache.stratos.load.balancer.conf.LoadBalancerConfiguration;
-import org.apache.stratos.load.balancer.conf.domain.MemberIpType;
import org.apache.stratos.load.balancer.conf.domain.TenantIdentifier;
import org.apache.stratos.load.balancer.context.LoadBalancerContext;
import org.apache.stratos.load.balancer.statistics.InFlightRequestDecrementCallable;
@@ -33,11 +37,7 @@ import org.apache.stratos.load.balancer.statistics.InFlightRequestIncrementCalla
import org.apache.stratos.load.balancer.statistics.LoadBalancerStatisticsExecutor;
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;
-import org.apache.stratos.messaging.domain.topology.Port;
import org.apache.stratos.messaging.message.receiver.tenant.TenantManager;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
import org.apache.synapse.MessageContext;
import org.apache.synapse.SynapseConstants;
import org.apache.synapse.SynapseException;
@@ -84,7 +84,9 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
public void init(SynapseEnvironment synapseEnvironment) {
super.init(synapseEnvironment);
- requestDelegator = new RequestDelegator(LoadBalanceAlgorithmFactory.createAlgorithm(algorithmClassName));
+ LoadBalanceAlgorithm algorithm = LoadBalanceAlgorithmFactory.createAlgorithm(algorithmClassName);
+ TopologyProvider topologyProvider = LoadBalancerConfiguration.getInstance().getTopologyProvider();
+ requestDelegator = new RequestDelegator(algorithm, topologyProvider);
synapseEnvironment.getSynapseConfiguration().setProperty(SynapseConstants.PROP_SAL_ENDPOINT_DEFAULT_SESSION_TIMEOUT, String.valueOf(sessionTimeout));
setDispatcher(new HttpSessionDispatcher());
}
@@ -252,7 +254,6 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
throwSynapseException(synCtx, 403, String.format("You are unauthorized to access"));
}
} else {
-
member = requestDelegator.findNextMemberFromHostName(targetHost, synCtx.getMessageID());
}
@@ -260,12 +261,24 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
return null;
// Create Axi2 member object
- org.apache.axis2.clustering.Member axis2Member = new org.apache.axis2.clustering.Member(
- getMemberIp(synCtx, member), -1);
+ return createAxis2Member(synCtx, targetHost, member);
+ }
+
+ /**
+ * Create axis2 member from load balancer member object.
+ * @param synCtx
+ * @param member
+ * @return
+ */
+ private org.apache.axis2.clustering.Member createAxis2Member(MessageContext synCtx, String clusterHostName,
+ Member member) {
+ org.apache.axis2.clustering.Member axis2Member =
+ new org.apache.axis2.clustering.Member(member.getHostName(), -1);
axis2Member.setDomain(member.getClusterId());
- axis2Member.setActive(member.isActive());
+ axis2Member.setActive(true);
// Set cluster id and member id in member properties
axis2Member.getProperties().setProperty(LoadBalancerConstants.CLUSTER_ID, member.getClusterId());
+ axis2Member.getProperties().setProperty(LoadBalancerConstants.CLUSTER_HOSTNAME, clusterHostName);
axis2Member.getProperties().setProperty(LoadBalancerConstants.MEMBER_ID, member.getMemberId());
// Update axis2 member ports
updateAxis2MemberPorts(synCtx, axis2Member);
@@ -364,7 +377,6 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
}
/**
->>>>>>> 400
* Find topology member from axis2 member using cluster id and member id defined in axis2 member properties.
*
* @param synCtx
@@ -372,35 +384,38 @@ 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(LoadBalancerConstants.CLUSTER_ID);
+ String hostName = axis2Member.getProperties().getProperty(LoadBalancerConstants.CLUSTER_HOSTNAME);
String memberId = axis2Member.getProperties().getProperty(LoadBalancerConstants.MEMBER_ID);
- if (StringUtils.isBlank(clusterId) || StringUtils.isBlank(memberId)) {
+ if (StringUtils.isBlank(hostName)) {
if (log.isErrorEnabled()) {
- log.error(String.format("Could not find cluster id and/or member id properties in axis2 member: [cluster-id] %s " +
- "[member-id] %s", clusterId, memberId));
+ log.error(String.format("Could not find hostname in axis2 member: [hostname] %s ", hostName));
}
throwSynapseException(synCtx, 500, "Internal server error");
}
- try {
- TopologyManager.acquireReadLock();
- Cluster cluster = LoadBalancerContext.getInstance().getClusterIdClusterMap().getCluster(clusterId);
- if (cluster == null) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Cluster not found in load balancer context: [cluster-id] %s ", clusterId));
- }
- throwSynapseException(synCtx, 500, "Internal server error");
+ if (StringUtils.isBlank(memberId)) {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Could not find member id in axis2 member: [member] %s ",
+ axis2Member.getHostName()));
}
- Member member = cluster.getMember(memberId);
- if (member == null) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Member not found in load balancer context: [cluster-id] %s [member-id] %s", clusterId, memberId));
- }
- throwSynapseException(synCtx, 500, "Internal server error");
+ throwSynapseException(synCtx, 500, "Internal server error");
+ }
+
+ Cluster cluster = requestDelegator.getCluster(hostName);
+ if (cluster == null) {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Cluster not found in load balancer context: [hostname] %s ", hostName));
}
- return member;
- } finally {
- TopologyManager.releaseReadLock();
+ throwSynapseException(synCtx, 500, "Internal server error");
+ }
+ Member member = cluster.getMember(memberId);
+ if (member == null) {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Member not found in load balancer context: [hostname] %s [member-id] %s",
+ hostName, memberId));
+ }
+ throwSynapseException(synCtx, 500, "Internal server error");
}
+ return member;
}
/**
@@ -410,34 +425,34 @@ public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints
* @param member
* @return
*/
- private String getMemberIp(MessageContext synCtx, Member member) {
- if (LoadBalancerConfiguration.getInstance().isTopologyEventListenerEnabled()) {
- if (LoadBalancerConfiguration.getInstance().getTopologyMemberIpType() == MemberIpType.Public) {
- // Return member's public IP address
- if (StringUtils.isBlank(member.getDefaultPublicIP())) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Member public IP address not found: [member] %s", member.getMemberId()));
- }
- throwSynapseException(synCtx, 500, "Internal server error");
- }
- if (log.isDebugEnabled()) {
- log.debug(String.format("Using member public IP address: [member] %s [ip] %s", member.getMemberId(), member.getDefaultPublicIP()));
- }
- return member.getDefaultPublicIP();
- }
- }
- // Return member's private IP address
- if (StringUtils.isBlank(member.getDefaultPrivateIP())) {
- if (log.isErrorEnabled()) {
- log.error(String.format("Member IP address not found: [member] %s", member.getMemberId()));
- }
- throwSynapseException(synCtx, 500, "Internal server error");
- }
- if (log.isDebugEnabled()) {
- log.debug(String.format("Using member IP address: [member] %s [ip] %s", member.getMemberId(), member.getDefaultPrivateIP()));
- }
- return member.getDefaultPrivateIP();
- }
+// private String getMemberIp(MessageContext synCtx, Member member) {
+// if (LoadBalancerConfiguration.getInstance().isTopologyEventListenerEnabled()) {
+// if (LoadBalancerConfiguration.getInstance().getTopologyMemberIpType() == MemberIpType.Public) {
+// // Return member's public IP address
+// if (StringUtils.isBlank(member.getDefaultPublicIP())) {
+// if (log.isErrorEnabled()) {
+// log.error(String.format("Member public IP address not found: [member] %s", member.getMemberId()));
+// }
+// throwSynapseException(synCtx, 500, "Internal server error");
+// }
+// if (log.isDebugEnabled()) {
+// log.debug(String.format("Using member public IP address: [member] %s [ip] %s", member.getMemberId(), member.getDefaultPublicIP()));
+// }
+// return member.getDefaultPublicIP();
+// }
+// }
+// // Return member's private IP address
+// if (StringUtils.isBlank(member.getDefaultPrivateIP())) {
+// if (log.isErrorEnabled()) {
+// log.error(String.format("Member IP address not found: [member] %s", member.getMemberId()));
+// }
+// throwSynapseException(synCtx, 500, "Internal server error");
+// }
+// if (log.isDebugEnabled()) {
+// log.debug(String.format("Using member IP address: [member] %s [ip] %s", member.getMemberId(), member.getDefaultPrivateIP()));
+// }
+// return member.getDefaultPrivateIP();
+// }
/**
* Extract incoming request URL from message context.
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/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 9196371..33ce789 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
@@ -26,23 +26,23 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.common.clustering.DistributedObjectProvider;
import org.apache.stratos.common.threading.StratosThreadPool;
+import org.apache.stratos.load.balancer.common.event.receivers.LoadBalancerApplicationSignUpEventReceiver;
+import org.apache.stratos.load.balancer.common.event.receivers.LoadBalancerDomainMappingEventReceiver;
+import org.apache.stratos.load.balancer.common.event.receivers.LoadBalancerTopologyEventReceiver;
import org.apache.stratos.load.balancer.common.statistics.notifier.LoadBalancerStatisticsNotifier;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
import org.apache.stratos.load.balancer.conf.LoadBalancerConfiguration;
import org.apache.stratos.load.balancer.conf.configurator.CEPConfigurator;
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.endpoint.EndpointDeployer;
import org.apache.stratos.load.balancer.exception.TenantAwareLoadBalanceEndpointException;
-import org.apache.stratos.load.balancer.messaging.receiver.LoadBalancerApplicationSignUpEventReceiver;
-import org.apache.stratos.load.balancer.messaging.receiver.LoadBalancerDomainMappingEventReceiver;
-import org.apache.stratos.load.balancer.messaging.receiver.LoadBalancerTenantEventReceiver;
-import org.apache.stratos.load.balancer.messaging.receiver.LoadBalancerTopologyEventReceiver;
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;
+import org.apache.stratos.messaging.message.receiver.tenant.TenantEventReceiver;
import org.apache.synapse.config.SynapseConfiguration;
import org.apache.synapse.config.xml.MultiXMLConfigurationBuilder;
import org.apache.synapse.core.SynapseEnvironment;
@@ -62,8 +62,7 @@ import org.wso2.carbon.utils.ConfigurationContextService;
import org.wso2.carbon.utils.multitenancy.MultitenantConstants;
import java.io.File;
-import java.util.Map;
-import java.util.Set;
+import java.util.Collection;
import java.util.concurrent.ExecutorService;
/**
@@ -93,7 +92,7 @@ public class LoadBalancerServiceComponent {
private boolean activated = false;
private ExecutorService executorService;
private LoadBalancerTopologyEventReceiver topologyReceiver;
- private LoadBalancerTenantEventReceiver tenantReceiver;
+ private TenantEventReceiver tenantReceiver;
private LoadBalancerApplicationSignUpEventReceiver applicationSignUpEventReceiver;
private LoadBalancerDomainMappingEventReceiver domainMappingEventReceiver;
private LoadBalancerStatisticsNotifier statisticsNotifier;
@@ -102,14 +101,12 @@ public class LoadBalancerServiceComponent {
try {
ClusteringAgent clusteringAgent = ServiceReferenceHolder.getInstance().getAxisConfiguration().getClusteringAgent();
boolean clusteringEnabled = (clusteringAgent != null);
- LoadBalancerContext.getInstance().setClustered(clusteringEnabled);
-
if(log.isInfoEnabled()) {
log.info(String.format("Load balancer clustering is %s", (clusteringEnabled ? "enabled" : "disabled")));
}
// Register endpoint deployer
- SynapseEnvironmentService synEnvService = LoadBalancerContext.getInstance().getTenantIdSynapseEnvironmentServiceMap()
+ SynapseEnvironmentService synEnvService = ServiceReferenceHolder.getInstance()
.getSynapseEnvironmentService(MultitenantConstants.SUPER_TENANT_ID);
registerDeployer(ServiceReferenceHolder.getInstance().getAxisConfiguration(),
synEnvService.getSynapseEnvironment());
@@ -129,22 +126,28 @@ public class LoadBalancerServiceComponent {
executorService = StratosThreadPool.getExecutorService(LoadBalancerConstants.LOAD_BALANCER_THREAD_POOL_ID,
threadPoolSize);
+ TopologyProvider topologyProvider = LoadBalancerConfiguration.getInstance().getTopologyProvider();
+ if(topologyProvider == null) {
+ topologyProvider = new TopologyProvider();
+ LoadBalancerConfiguration.getInstance().setTopologyProvider(topologyProvider);
+ }
+
if (configuration.isMultiTenancyEnabled()) {
// Start tenant event receiver
- startTenantEventReceiver(executorService);
+ startTenantEventReceiver(executorService, topologyProvider);
}
if(configuration.isDomainMappingEnabled()) {
// Start application signup event receiver
- startApplicationSignUpEventReceiver(executorService);
+ startApplicationSignUpEventReceiver(executorService, topologyProvider);
// Start domain mapping event receiver
- startDomainMappingEventReceiver(executorService);
+ startDomainMappingEventReceiver(executorService, topologyProvider);
}
if (configuration.isTopologyEventListenerEnabled()) {
// Start topology receiver
- startTopologyEventReceiver(executorService);
+ startTopologyEventReceiver(executorService, topologyProvider);
if (log.isInfoEnabled()) {
if (TopologyServiceFilter.getInstance().isActive()) {
@@ -197,8 +200,9 @@ public class LoadBalancerServiceComponent {
}
}
- private void startTenantEventReceiver(ExecutorService executorService) {
- tenantReceiver = new LoadBalancerTenantEventReceiver();
+ private void startTenantEventReceiver(ExecutorService executorService, TopologyProvider topologyProvider) {
+
+ tenantReceiver = new TenantEventReceiver();
tenantReceiver.setExecutorService(executorService);
tenantReceiver.execute();
if (log.isInfoEnabled()) {
@@ -206,8 +210,8 @@ public class LoadBalancerServiceComponent {
}
}
- private void startDomainMappingEventReceiver(ExecutorService executorService) {
- domainMappingEventReceiver = new LoadBalancerDomainMappingEventReceiver();
+ private void startDomainMappingEventReceiver(ExecutorService executorService, TopologyProvider topologyProvider) {
+ domainMappingEventReceiver = new LoadBalancerDomainMappingEventReceiver(topologyProvider);
domainMappingEventReceiver.setExecutorService(executorService);
domainMappingEventReceiver.execute();
if (log.isInfoEnabled()) {
@@ -215,8 +219,8 @@ public class LoadBalancerServiceComponent {
}
}
- private void startApplicationSignUpEventReceiver(ExecutorService executorService) {
- applicationSignUpEventReceiver = new LoadBalancerApplicationSignUpEventReceiver();
+ private void startApplicationSignUpEventReceiver(ExecutorService executorService, TopologyProvider topologyProvider) {
+ applicationSignUpEventReceiver = new LoadBalancerApplicationSignUpEventReceiver(topologyProvider);
applicationSignUpEventReceiver.setExecutorService(executorService);
applicationSignUpEventReceiver.execute();
if (log.isInfoEnabled()) {
@@ -224,8 +228,9 @@ public class LoadBalancerServiceComponent {
}
}
- private void startTopologyEventReceiver(ExecutorService executorService) {
- topologyReceiver = new LoadBalancerTopologyEventReceiver();
+ private void startTopologyEventReceiver(ExecutorService executorService, TopologyProvider topologyProvider) {
+
+ topologyReceiver = new LoadBalancerTopologyEventReceiver(topologyProvider);
topologyReceiver.setExecutorService(executorService);
topologyReceiver.execute();
if (log.isInfoEnabled()) {
@@ -278,36 +283,41 @@ public class LoadBalancerServiceComponent {
protected void deactivate(ComponentContext context) {
try {
- Set<Map.Entry<Integer, SynapseEnvironmentService>> entrySet = LoadBalancerContext
- .getInstance().getTenantIdSynapseEnvironmentServiceMap().getTenantIdSynapseEnvironmentServiceMap().entrySet();
- for (Map.Entry<Integer, SynapseEnvironmentService> entry : entrySet) {
- unregisterDeployer(entry.getValue().getConfigurationContext()
- .getAxisConfiguration(), entry.getValue()
- .getSynapseEnvironment());
+ Collection<SynapseEnvironmentService> synapseEnvironmentServices =
+ ServiceReferenceHolder.getInstance().getSynapseEnvironmentServices();
+ for (SynapseEnvironmentService synapseEnvironmentService : synapseEnvironmentServices) {
+ unregisterDeployer(synapseEnvironmentService.getConfigurationContext()
+ .getAxisConfiguration(), synapseEnvironmentService.getSynapseEnvironment());
}
} catch (Exception e) {
log.warn("An error occurred while removing endpoint deployer", e);
}
// Terminate tenant receiver
- try {
- tenantReceiver.terminate();
- } catch (Exception e) {
- log.warn("An error occurred while terminating tenant event receiver", e);
+ if(tenantReceiver != null) {
+ try {
+ tenantReceiver.terminate();
+ } catch (Exception e) {
+ log.warn("An error occurred while terminating tenant event receiver", e);
+ }
}
// Terminate topology receiver
- try {
- topologyReceiver.terminate();
- } catch (Exception e) {
- log.warn("An error occurred while terminating topology event receiver", e);
+ if(topologyReceiver != null) {
+ 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);
+ if(statisticsNotifier != null) {
+ try {
+ statisticsNotifier.terminate();
+ } catch (Exception e) {
+ log.warn("An error occurred while terminating health statistics notifier", e);
+ }
}
// Shutdown executor service
@@ -402,13 +412,11 @@ public class LoadBalancerServiceComponent {
* new Synapse Instance
*/
protected void setSynapseEnvironmentService(SynapseEnvironmentService synapseEnvironmentService) {
- boolean alreadyCreated = LoadBalancerContext.getInstance()
- .getTenantIdSynapseEnvironmentServiceMap()
+ boolean alreadyCreated = ServiceReferenceHolder.getInstance()
.containsSynapseEnvironmentService(synapseEnvironmentService.getTenantId());
- LoadBalancerContext.getInstance().getTenantIdSynapseEnvironmentServiceMap()
- .addSynapseEnvironmentService(synapseEnvironmentService.getTenantId(),
- synapseEnvironmentService);
+ ServiceReferenceHolder.getInstance().addSynapseEnvironmentService(synapseEnvironmentService.getTenantId(),
+ synapseEnvironmentService);
if (activated) {
if (!alreadyCreated) {
try {
@@ -431,10 +439,8 @@ public class LoadBalancerServiceComponent {
*
* @param synapseEnvironmentService synapseEnvironment
*/
- protected void unsetSynapseEnvironmentService(
- SynapseEnvironmentService synapseEnvironmentService) {
- LoadBalancerContext.getInstance().getTenantIdSynapseEnvironmentServiceMap().removeSynapseEnvironmentService(
- synapseEnvironmentService.getTenantId());
+ protected void unsetSynapseEnvironmentService(SynapseEnvironmentService synapseEnvironmentService) {
+ ServiceReferenceHolder.getInstance().removeSynapseEnvironmentService(synapseEnvironmentService.getTenantId());
}
protected void setRegistryService(RegistryService regService) {
@@ -481,16 +487,12 @@ public class LoadBalancerServiceComponent {
protected void unsetSynapseRegistrationsService(
SynapseRegistrationsService synapseRegistrationsService) {
int tenantId = synapseRegistrationsService.getTenantId();
- if (LoadBalancerContext.getInstance().getTenantIdSynapseEnvironmentServiceMap()
- .containsSynapseEnvironmentService(tenantId)) {
- SynapseEnvironment env = LoadBalancerContext.getInstance()
- .getTenantIdSynapseEnvironmentServiceMap()
- .getSynapseEnvironmentService(tenantId)
+ if (ServiceReferenceHolder.getInstance().containsSynapseEnvironmentService(tenantId)) {
+ SynapseEnvironment env = ServiceReferenceHolder.getInstance().getSynapseEnvironmentService(tenantId)
.getSynapseEnvironment();
- LoadBalancerContext.getInstance().getTenantIdSynapseEnvironmentServiceMap()
- .removeSynapseEnvironmentService(
- synapseRegistrationsService.getTenantId());
+ ServiceReferenceHolder.getInstance().removeSynapseEnvironmentService(
+ synapseRegistrationsService.getTenantId());
AxisConfiguration axisConfig = synapseRegistrationsService
.getConfigurationContext().getAxisConfiguration();
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/ServiceReferenceHolder.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/ServiceReferenceHolder.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/ServiceReferenceHolder.java
index 54e9f25..e026ac5 100644
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/ServiceReferenceHolder.java
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/ServiceReferenceHolder.java
@@ -27,9 +27,13 @@ import org.apache.stratos.common.clustering.DistributedObjectProvider;
import org.apache.stratos.load.balancer.exception.TenantAwareLoadBalanceEndpointException;
import org.apache.synapse.config.SynapseConfiguration;
import org.wso2.carbon.mediation.dependency.mgt.services.DependencyManagementService;
+import org.wso2.carbon.mediation.initializer.services.SynapseEnvironmentService;
import org.wso2.carbon.registry.core.session.UserRegistry;
import org.wso2.carbon.user.core.service.RealmService;
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+
/**
* Service reference holder keeps references of bind OSGi services.
*/
@@ -46,8 +50,10 @@ public class ServiceReferenceHolder {
private DependencyManagementService dependencyManager;
private RealmService realmService;
private DistributedObjectProvider distributedObjectProvider;
+ private ConcurrentHashMap<Integer, SynapseEnvironmentService> synapseEnvironmentServiceMap;
private ServiceReferenceHolder() {
+ synapseEnvironmentServiceMap = new ConcurrentHashMap<Integer, SynapseEnvironmentService>();
}
public static ServiceReferenceHolder getInstance() {
@@ -135,4 +141,24 @@ public class ServiceReferenceHolder {
public DistributedObjectProvider getDistributedObjectProvider() {
return distributedObjectProvider;
}
+
+ public SynapseEnvironmentService getSynapseEnvironmentService(int tenantId) {
+ return synapseEnvironmentServiceMap.get(tenantId);
+ }
+
+ public boolean containsSynapseEnvironmentService(int tenantId) {
+ return synapseEnvironmentServiceMap.containsKey(tenantId);
+ }
+
+ public void addSynapseEnvironmentService(int tenantId, SynapseEnvironmentService synapseEnvironmentService) {
+ synapseEnvironmentServiceMap.put(tenantId, synapseEnvironmentService);
+ }
+
+ public void removeSynapseEnvironmentService(int tenantId) {
+ synapseEnvironmentServiceMap.remove(tenantId);
+ }
+
+ public Collection<SynapseEnvironmentService> getSynapseEnvironmentServices() {
+ return synapseEnvironmentServiceMap.values();
+ }
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/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 88ec7bf..c8d0c2c 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
@@ -19,6 +19,7 @@
package org.apache.stratos.load.balancer.mediators;
import org.apache.commons.lang3.StringUtils;
+import org.apache.stratos.load.balancer.common.topology.TopologyProvider;
import org.apache.stratos.load.balancer.conf.LoadBalancerConfiguration;
import org.apache.stratos.load.balancer.context.LoadBalancerContext;
import org.apache.stratos.load.balancer.util.LoadBalancerConstants;
@@ -66,8 +67,8 @@ public class LocationReWriter extends AbstractMediator {
// Check whether the location host is an ip address of a known member
String hostname = LoadBalancerContext.getInstance().getMemberIpHostnameMap().get(inLocationUrl.getHost());
if (StringUtils.isEmpty(hostname)) {
-
- if (!LoadBalancerContext.getInstance().getHostNameClusterMap().containsCluster(inLocationUrl.getHost())) {
+ TopologyProvider topologyProvider = LoadBalancerConfiguration.getInstance().getTopologyProvider();
+ if (topologyProvider.getClusterByHostName(inLocationUrl.getHost()) == null) {
if (log.isDebugEnabled()) {
log.debug(String.format("A hostname not found for ip: [ip-address] %s", inLocationUrl.getHost()));
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerApplicationSignUpEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerApplicationSignUpEventReceiver.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerApplicationSignUpEventReceiver.java
deleted file mode 100644
index 37a0c97..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerApplicationSignUpEventReceiver.java
+++ /dev/null
@@ -1,69 +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.messaging.receiver;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.load.balancer.context.LoadBalancerContextUtil;
-import org.apache.stratos.messaging.domain.application.signup.ApplicationSignUp;
-import org.apache.stratos.messaging.domain.application.signup.DomainMapping;
-import org.apache.stratos.messaging.event.Event;
-import org.apache.stratos.messaging.event.application.signup.CompleteApplicationSignUpsEvent;
-import org.apache.stratos.messaging.listener.application.signup.CompleteApplicationSignUpsEventListener;
-import org.apache.stratos.messaging.message.receiver.application.signup.ApplicationSignUpEventReceiver;
-
-/**
- * Load balancer application signup event receiver.
- */
-public class LoadBalancerApplicationSignUpEventReceiver extends ApplicationSignUpEventReceiver {
-
- private static final Log log = LogFactory.getLog(LoadBalancerApplicationSignUpEventReceiver.class);
-
- public LoadBalancerApplicationSignUpEventReceiver() {
- addEventListeners();
- }
-
- private void addEventListeners() {
- addEventListener(new CompleteApplicationSignUpsEventListener() {
- @Override
- protected void onEvent(Event event) {
- if (log.isDebugEnabled()) {
- log.debug("Complete application signup event received");
- }
- CompleteApplicationSignUpsEvent completeApplicationSignUpsEvent = (CompleteApplicationSignUpsEvent)event;
- for(ApplicationSignUp applicationSignUp : completeApplicationSignUpsEvent.getApplicationSignUps()) {
- if(applicationSignUp.getDomainMappings() != null) {
- for (DomainMapping domainMapping : applicationSignUp.getDomainMappings()) {
- if(domainMapping != null) {
- LoadBalancerContextUtil.addClusterAgainstDomain(
- domainMapping.getServiceName(),
- domainMapping.getClusterId(),
- domainMapping.getDomainName());
-
- LoadBalancerContextUtil.addContextPathAgainstDomain(domainMapping.getDomainName(),
- domainMapping.getContextPath());
- }
- }
- }
- }
- }
- });
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c799abce/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerDomainMappingEventReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerDomainMappingEventReceiver.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerDomainMappingEventReceiver.java
deleted file mode 100644
index ea9238f..0000000
--- a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/messaging/receiver/LoadBalancerDomainMappingEventReceiver.java
+++ /dev/null
@@ -1,72 +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.messaging.receiver;
-
-import org.apache.stratos.load.balancer.context.LoadBalancerContextUtil;
-import org.apache.stratos.messaging.event.Event;
-import org.apache.stratos.messaging.event.domain.mapping.DomainMappingAddedEvent;
-import org.apache.stratos.messaging.event.domain.mapping.DomainMappingRemovedEvent;
-import org.apache.stratos.messaging.listener.domain.mapping.DomainMappingAddedEventListener;
-import org.apache.stratos.messaging.listener.domain.mapping.DomainMappingRemovedEventListener;
-import org.apache.stratos.messaging.message.receiver.domain.mapping.DomainMappingEventReceiver;
-
-/**
- * Load balancer domain mapping event receiver.
- */
-public class LoadBalancerDomainMappingEventReceiver extends DomainMappingEventReceiver {
-
- public LoadBalancerDomainMappingEventReceiver() {
- addEventListeners();
- }
-
- private void addEventListeners() {
-
- addEventListener(new DomainMappingAddedEventListener() {
- @Override
- protected void onEvent(Event event) {
- DomainMappingAddedEvent domainMappingAddedEvent = (DomainMappingAddedEvent)event;
-
- LoadBalancerContextUtil.addClusterAgainstDomain(
- domainMappingAddedEvent.getServiceName(),
- domainMappingAddedEvent.getClusterId(),
- domainMappingAddedEvent.getDomainName());
-
- LoadBalancerContextUtil.addContextPathAgainstDomain(
- domainMappingAddedEvent.getDomainName(),
- domainMappingAddedEvent.getContextPath());
- }
- });
-
- addEventListener(new DomainMappingRemovedEventListener() {
- @Override
- protected void onEvent(Event event) {
- DomainMappingRemovedEvent domainMappingRemovedEvent = (DomainMappingRemovedEvent)event;
-
- LoadBalancerContextUtil.removeClusterAgainstDomain(
- domainMappingRemovedEvent.getServiceName(),
- domainMappingRemovedEvent.getClusterId(),
- domainMappingRemovedEvent.getDomainName());
-
- LoadBalancerContextUtil.removeContextPathAgainstDomain(
- domainMappingRemovedEvent.getDomainName());
- }
- });
- }
-}