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/02/23 11:05:39 UTC
[02/11] stratos git commit: JCA - Removed unused code in extensions
managment.
JCA - Removed unused code in extensions managment.
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/1a8e7a64
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/1a8e7a64
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/1a8e7a64
Branch: refs/heads/master
Commit: 1a8e7a640e0a98ba8caa74a4e06ae52bbd6344c6
Parents: 8ef8e65
Author: Chamila de Alwis <ch...@wso2.com>
Authored: Mon Feb 23 01:45:20 2015 +0530
Committer: Imesh Gunaratne <im...@apache.org>
Committed: Mon Feb 23 15:35:34 2015 +0530
----------------------------------------------------------------------
.../org.apache.stratos.cartridge.agent/pom.xml | 39 --
.../extensions/DefaultExtensionHandler.java | 622 ++-----------------
.../cartridge/agent/util/ExtensionUtils.java | 108 ++--
3 files changed, 76 insertions(+), 693 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/1a8e7a64/components/org.apache.stratos.cartridge.agent/pom.xml
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/pom.xml b/components/org.apache.stratos.cartridge.agent/pom.xml
index a6cf765..a4f5d99 100644
--- a/components/org.apache.stratos.cartridge.agent/pom.xml
+++ b/components/org.apache.stratos.cartridge.agent/pom.xml
@@ -33,45 +33,6 @@
<description>Apache Stratos Cartridge Agent</description>
<url>http://apache.org</url>
- <profiles>
- <profile>
- <id>default</id>
- <activation>
- <activeByDefault>true</activeByDefault>
- </activation>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <excludes>
- <exclude>**/*/JavaCartridgeAgentTest.java</exclude>
- </excludes>
- </configuration>
- </plugin>
- </plugins>
- </build>
- </profile>
- <profile>
- <id>live</id>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <includes>
- <include>**/*/JavaCartridgeAgentTest.java</include>
- </includes>
- </configuration>
- </plugin>
- </plugins>
- </build>
- </profile>
- </profiles>
-
-
<dependencies>
<dependency>
<groupId>net.lingala.zip4j</groupId>
http://git-wip-us.apache.org/repos/asf/stratos/blob/1a8e7a64/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/extensions/DefaultExtensionHandler.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/extensions/DefaultExtensionHandler.java b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/extensions/DefaultExtensionHandler.java
index 20a18dd..8112d47 100644
--- a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/extensions/DefaultExtensionHandler.java
+++ b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/extensions/DefaultExtensionHandler.java
@@ -45,7 +45,6 @@ import org.apache.stratos.messaging.event.tenant.TenantSubscribedEvent;
import org.apache.stratos.messaging.event.tenant.TenantUnSubscribedEvent;
import org.apache.stratos.messaging.event.topology.*;
import org.apache.stratos.messaging.message.receiver.tenant.TenantManager;
-import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
import java.lang.reflect.Type;
import java.util.*;
@@ -60,7 +59,6 @@ public class DefaultExtensionHandler implements ExtensionHandler {
}.getType();
private static final Type serviceType = new TypeToken<Collection<Service>>() {
}.getType();
- private final ArrayList<Member> wkMembers = new ArrayList<Member>();
@Override
public void onInstanceStartedEvent() {
@@ -112,6 +110,7 @@ public class DefaultExtensionHandler implements ExtensionHandler {
if (log.isInfoEnabled()) {
log.info("Executing git checkout");
}
+
RepositoryInformation repoInformation = new RepositoryInformation();
repoInformation.setRepoUsername(repoUsername);
if (repoPassword == null) {
@@ -119,6 +118,7 @@ public class DefaultExtensionHandler implements ExtensionHandler {
} else {
repoInformation.setRepoPassword(repoPassword);
}
+
repoInformation.setRepoUrl(repoURL);
repoInformation.setRepoPath(localRepoPath);
repoInformation.setTenantId(tenantId);
@@ -129,6 +129,7 @@ public class DefaultExtensionHandler implements ExtensionHandler {
} catch (Exception e) {
log.error(e);
}
+
Map<String, String> env = new HashMap<String, String>();
env.put("STRATOS_ARTIFACT_UPDATED_CLUSTER_ID", artifactUpdatedEvent.getClusterId());
env.put("STRATOS_ARTIFACT_UPDATED_TENANT_ID", artifactUpdatedEvent.getTenantId());
@@ -222,7 +223,6 @@ public class DefaultExtensionHandler implements ExtensionHandler {
CartridgeAgentEventPublisher.publishInstanceReadyToShutdownEvent();
}
-
@Override
public void onMemberActivatedEvent(MemberActivatedEvent memberActivatedEvent) {
if (log.isInfoEnabled()) {
@@ -235,92 +235,16 @@ public class DefaultExtensionHandler implements ExtensionHandler {
log.debug("Member activated event msg:" + msg);
}
- boolean isConsistent = ExtensionUtils.checkTopologyConsistency(memberActivatedEvent.getServiceName(),
+ boolean memberInitialized = ExtensionUtils.checkTopologyConsistency(memberActivatedEvent.getServiceName(),
memberActivatedEvent.getClusterId(), memberActivatedEvent.getMemberId());
- if (!isConsistent) {
+ if (!memberInitialized) {
if (log.isErrorEnabled()) {
- log.error("Topology is inconsistent...failed to execute member activated event");
+ log.error("Member has not initialized. Failed to execute member activated event");
}
return;
}
- Topology topology = TopologyManager.getTopology();
- Service service = topology.getService(memberActivatedEvent.getServiceName());
- Cluster cluster = service.getCluster(memberActivatedEvent.getClusterId());
- String lbClusterId = cluster.getMember(memberActivatedEvent.getMemberId()).getLbClusterId();
- Member member = cluster.getMember(memberActivatedEvent.getMemberId());
-
- // check whether member activated event is received from the same cluster, lbcluster or cartridge group
- if (ExtensionUtils.isRelevantMemberEvent(memberActivatedEvent.getServiceName(), memberActivatedEvent.getClusterId(), lbClusterId)) {
- Map<String, String> env = new HashMap<String, String>();
- env.put("STRATOS_MEMBER_ACTIVATED_MEMBER_IP", memberActivatedEvent.getDefaultPrivateIP());
- env.put("STRATOS_MEMBER_ACTIVATED_MEMBER_ID", memberActivatedEvent.getMemberId());
- env.put("STRATOS_MEMBER_ACTIVATED_CLUSTER_ID", memberActivatedEvent.getClusterId());
- env.put("STRATOS_MEMBER_ACTIVATED_LB_CLUSTER_ID", lbClusterId);
- env.put("STRATOS_MEMBER_ACTIVATED_NETWORK_PARTITION_ID", memberActivatedEvent.getNetworkPartitionId());
- env.put("STRATOS_MEMBER_ACTIVATED_SERVICE_NAME", memberActivatedEvent.getServiceName());
-
- Collection<Port> ports = memberActivatedEvent.getPorts();
- String ports_str = "";
- for (Port port : ports) {
- ports_str += port.getProtocol() + "," + port.getValue() + "," + port.getProxy() + "|";
- }
- env.put("STRATOS_MEMBER_ACTIVATED_PORTS", ports_str);
-
- Collection<Member> members = cluster.getMembers();
- env.put("STRATOS_MEMBER_ACTIVATED_MEMBER_LIST_JSON", gson.toJson(members, memberType));
- String[] memberIps = ExtensionUtils.getLbMemberIp(lbClusterId);
- if (memberIps != null && memberIps.length > 1) {
- env.put("STRATOS_MEMBER_ACTIVATED_LB_IP", memberIps[0]);
- env.put("STRATOS_MEMBER_ACTIVATED_LB_PUBLIC_IP", memberIps[1]);
- }
- env.put("STRATOS_TOPOLOGY_JSON", gson.toJson(topology.getServices(), serviceType));
- ExtensionUtils.addProperties(service.getProperties(), env, "MEMBER_ACTIVATED_SERVICE_PROPERTY");
- ExtensionUtils.addProperties(cluster.getProperties(), env, "MEMBER_ACTIVATED_CLUSTER_PROPERTY");
- ExtensionUtils.addProperties(member.getProperties(), env, "MEMBER_ACTIVATED_MEMBER_PROPERTY");
-
- // if clustering is enabled check activated member is WK member
- String flagClustering = CartridgeAgentConfiguration.getInstance().getIsClustered();
-
- // if WK member is re-spawned, update axis2.xml
- if (member.getProperties() != null && "true".equals(member.getProperties().getProperty(CartridgeAgentConstants.CLUSTERING_PRIMARY_KEY)) &&
- flagClustering != null && "true".equals(flagClustering.toLowerCase())) {
- if (log.isDebugEnabled()) {
- log.debug(" If WK member is re-spawned, update axis2.xml ");
- }
- boolean hasWKIpChanged = true;
- for (Member m : this.wkMembers) {
- if (m.getDefaultPrivateIP().equals(memberActivatedEvent.getDefaultPrivateIP())) {
- hasWKIpChanged = false;
- }
- }
- if (log.isDebugEnabled()) {
- log.debug(" hasWKIpChanged " + hasWKIpChanged);
- }
- int minCount = Integer.parseInt(CartridgeAgentConfiguration.getInstance().getMinCount());
- boolean isWKMemberGroupReady = isWKMemberGroupReady(env, minCount);
- if (log.isDebugEnabled()) {
- log.debug("minCount " + minCount);
- log.debug("isWKMemberGroupReady " + isWKMemberGroupReady);
- }
- if (hasWKIpChanged && isWKMemberGroupReady) {
- if (log.isDebugEnabled()) {
- log.debug("Setting env var STRATOS_UPDATE_WK_IP to true");
- }
- env.put("STRATOS_UPDATE_WK_IP", "true");
- }
- }
- if (log.isDebugEnabled()) {
- log.debug("Setting env var STRATOS_CLUSTERING to " + flagClustering);
- }
- env.put("STRATOS_CLUSTERING", flagClustering);
- env.put("STRATOS_WK_MEMBER_COUNT", CartridgeAgentConfiguration.getInstance().getMinCount());
- ExtensionUtils.executeMemberActivatedExtension(env);
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Member activated event is not relevant...skipping agent extension");
- }
- }
+ ExtensionUtils.executeMemberActivatedExtension(Collections.<String, String>emptyMap());
}
@Override
@@ -352,7 +276,6 @@ public class DefaultExtensionHandler implements ExtensionHandler {
ExtensionUtils.executeCompleteTopologyExtension(env);
}
-
@Override
public void onMemberInitializedEvent(MemberInitializedEvent memberInitializedEvent) {
if (log.isDebugEnabled()) {
@@ -362,27 +285,24 @@ public class DefaultExtensionHandler implements ExtensionHandler {
String clusterIdInPayload = CartridgeAgentConfiguration.getInstance().getClusterId();
String memberIdInPayload = CartridgeAgentConfiguration.getInstance().getMemberId();
-
- boolean isConsistent = ExtensionUtils.checkTopologyConsistency(serviceNameInPayload, clusterIdInPayload, memberIdInPayload);
- if (!isConsistent) {
+// boolean memberInitialized = ExtensionUtils.checkTopologyConsistency(serviceNameInPayload, clusterIdInPayload, memberIdInPayload);
+ boolean memberExists = ExtensionUtils.memberExistsInTopology(serviceNameInPayload, clusterIdInPayload, memberIdInPayload);
+ if (!memberExists) {
// if this member isn't there in the complete topology
if (log.isDebugEnabled()) {
- log.debug("Member initialized event, topology is inconsistent");
+ log.debug("Member does not exist in topology, or not in initialized state.");
}
- return;
} else {
if (log.isDebugEnabled()) {
- log.debug("Member initialized event, topology is consistent, agent is initialized");
+ log.debug("Member has initialized, topology is consistent, agent is initialized");
}
+
CartridgeAgentConfiguration.getInstance().setInitialized(true);
}
-
}
-
@Override
public void onMemberCreatedEvent(MemberCreatedEvent memberCreatedEvent) {
-
// listen to this just to get updated faster about the member initialization
if (log.isDebugEnabled()) {
log.debug("Instance Spawned event received");
@@ -390,11 +310,9 @@ public class DefaultExtensionHandler implements ExtensionHandler {
String serviceNameInPayload = CartridgeAgentConfiguration.getInstance().getServiceName();
String clusterIdInPayload = CartridgeAgentConfiguration.getInstance().getClusterId();
String memberIdInPayload = CartridgeAgentConfiguration.getInstance().getMemberId();
- boolean isConsistent = ExtensionUtils.checkTopologyConsistency(serviceNameInPayload, clusterIdInPayload, memberIdInPayload);
- if (!isConsistent) {
- // if this event is not relevant to this member
- return;
- } else {
+
+ boolean memberInitialized = ExtensionUtils.checkTopologyConsistency(serviceNameInPayload, clusterIdInPayload, memberIdInPayload);
+ if (memberInitialized) {
CartridgeAgentConfiguration.getInstance().setInitialized(true);
}
}
@@ -425,50 +343,16 @@ public class DefaultExtensionHandler implements ExtensionHandler {
log.debug("Member terminated event msg:" + msg);
}
- boolean isConsistent = ExtensionUtils.checkTopologyConsistency(memberTerminatedEvent.getServiceName(),
+ boolean memberInitialized = ExtensionUtils.checkTopologyConsistency(memberTerminatedEvent.getServiceName(),
memberTerminatedEvent.getClusterId(), memberTerminatedEvent.getMemberId());
- if (!isConsistent) {
+ if (!memberInitialized) {
if (log.isErrorEnabled()) {
- log.error("Topology is inconsistent...failed to execute member terminated event");
+ log.error("Member has not initialized. Failed to execute member terminated event");
}
return;
}
- Topology topology = TopologyManager.getTopology();
- Service service = topology.getService(memberTerminatedEvent.getServiceName());
- Cluster cluster = service.getCluster(memberTerminatedEvent.getClusterId());
- Member terminatedMember = cluster.getMember(memberTerminatedEvent.getMemberId());
- String lbClusterId = cluster.getMember(memberTerminatedEvent.getClusterId()).getLbClusterId();
-
- // check whether terminated member is within the same cluster, LB cluster or cartridge group
- if (ExtensionUtils.isRelevantMemberEvent(memberTerminatedEvent.getServiceName(),
- memberTerminatedEvent.getClusterId(), lbClusterId)) {
-
- Collection<Member> members = cluster.getMembers();
- Map<String, String> env = new HashMap<String, String>();
- env.put("STRATOS_MEMBER_TERMINATED_MEMBER_IP", terminatedMember.getDefaultPrivateIP());
- env.put("STRATOS_MEMBER_TERMINATED_MEMBER_ID", memberTerminatedEvent.getMemberId());
- env.put("STRATOS_MEMBER_TERMINATED_CLUSTER_ID", memberTerminatedEvent.getClusterId());
- env.put("STRATOS_MEMBER_TERMINATED_LB_CLUSTER_ID", lbClusterId);
- env.put("STRATOS_MEMBER_TERMINATED_NETWORK_PARTITION_ID", memberTerminatedEvent.getNetworkPartitionId());
- env.put("STRATOS_MEMBER_TERMINATED_SERVICE_NAME", memberTerminatedEvent.getServiceName());
- env.put("STRATOS_MEMBER_TERMINATED_MEMBER_LIST_JSON", gson.toJson(members, memberType));
- env.put("STRATOS_TOPOLOGY_JSON", gson.toJson(topology.getServices(), serviceType));
- String[] memberIps = ExtensionUtils.getLbMemberIp(lbClusterId);
- if (memberIps != null && memberIps.length > 1) {
- env.put("STRATOS_MEMBER_TERMINATED_LB_IP", memberIps[0]);
- env.put("STRATOS_MEMBER_TERMINATED_LB_PUBLIC_IP", memberIps[1]);
- }
-
- ExtensionUtils.addProperties(service.getProperties(), env, "MEMBER_TERMINATED_SERVICE_PROPERTY");
- ExtensionUtils.addProperties(cluster.getProperties(), env, "MEMBER_TERMINATED_CLUSTER_PROPERTY");
- ExtensionUtils.addProperties(terminatedMember.getProperties(), env, "MEMBER_TERMINATED_MEMBER_PROPERTY");
- ExtensionUtils.executeMemberTerminatedExtension(env);
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Member terminated event is not relevant...skipping agent extension");
- }
- }
+ ExtensionUtils.executeMemberTerminatedExtension(Collections.<String, String>emptyMap());
}
@Override
@@ -483,49 +367,16 @@ public class DefaultExtensionHandler implements ExtensionHandler {
log.debug("Member suspended event msg:" + msg);
}
- boolean isConsistent = ExtensionUtils.checkTopologyConsistency(memberSuspendedEvent.getServiceName(),
+ boolean memberInitialized = ExtensionUtils.checkTopologyConsistency(memberSuspendedEvent.getServiceName(),
memberSuspendedEvent.getClusterId(), memberSuspendedEvent.getMemberId());
- if (!isConsistent) {
+ if (!memberInitialized) {
if (log.isErrorEnabled()) {
- log.error("Topology is inconsistent...failed to execute member suspended event");
+ log.error("Member has not initialized. Failed to execute member suspended event");
}
return;
}
- String clusterId = memberSuspendedEvent.getClusterId();
- Topology topology = TopologyManager.getTopology();
- Service service = topology.getService(memberSuspendedEvent.getServiceName());
- Cluster cluster = service.getCluster(memberSuspendedEvent.getClusterId());
- Member suspendedMember = cluster.getMember(memberSuspendedEvent.getMemberId());
- String lbClusterId = cluster.getMember(memberSuspendedEvent.getClusterId()).getLbClusterId();
-
- // check whether new member is in the same member cluster or LB cluster of this instance
- if (ExtensionUtils.isRelevantMemberEvent(memberSuspendedEvent.getServiceName(),
- memberSuspendedEvent.getClusterId(), lbClusterId)) {
- Collection<Member> members = cluster.getMembers();
- Map<String, String> env = new HashMap<String, String>();
- env.put("STRATOS_MEMBER_SUSPENDED_MEMBER_IP", suspendedMember.getDefaultPrivateIP());
- env.put("STRATOS_MEMBER_SUSPENDED_MEMBER_ID", memberSuspendedEvent.getMemberId());
- env.put("STRATOS_MEMBER_SUSPENDED_CLUSTER_ID", memberSuspendedEvent.getClusterId());
- env.put("STRATOS_MEMBER_SUSPENDED_LB_CLUSTER_ID", lbClusterId);
- env.put("STRATOS_MEMBER_SUSPENDED_NETWORK_PARTITION_ID", memberSuspendedEvent.getNetworkPartitionId());
- env.put("STRATOS_MEMBER_SUSPENDED_SERVICE_NAME", memberSuspendedEvent.getServiceName());
- env.put("STRATOS_MEMBER_SUSPENDED_MEMBER_LIST_JSON", gson.toJson(members, memberType));
- env.put("STRATOS_TOPOLOGY_JSON", gson.toJson(topology.getServices(), serviceType));
- String[] memberIps = ExtensionUtils.getLbMemberIp(lbClusterId);
- if (memberIps != null && memberIps.length > 1) {
- env.put("STRATOS_MEMBER_SUSPENDED_LB_IP", memberIps[0]);
- env.put("STRATOS_MEMBER_SUSPENDED_LB_PUBLIC_IP", memberIps[1]);
- }
- ExtensionUtils.addProperties(service.getProperties(), env, "MEMBER_SUSPENDED_SERVICE_PROPERTY");
- ExtensionUtils.addProperties(cluster.getProperties(), env, "MEMBER_SUSPENDED_CLUSTER_PROPERTY");
- ExtensionUtils.addProperties(suspendedMember.getProperties(), env, "MEMBER_SUSPENDED_MEMBER_PROPERTY");
- ExtensionUtils.executeMemberSuspendedExtension(env);
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Member suspended event is not relevant...skipping agent extension");
- }
- }
+ ExtensionUtils.executeMemberSuspendedExtension(Collections.<String, String>emptyMap());
}
@Override
@@ -540,373 +391,16 @@ public class DefaultExtensionHandler implements ExtensionHandler {
log.debug("Member started event msg:" + msg);
}
- boolean isConsistent = ExtensionUtils.checkTopologyConsistency(memberStartedEvent.getServiceName(),
+ boolean memberInitialized = ExtensionUtils.checkTopologyConsistency(memberStartedEvent.getServiceName(),
memberStartedEvent.getClusterId(), memberStartedEvent.getMemberId());
- if (!isConsistent) {
+ if (!memberInitialized) {
if (log.isErrorEnabled()) {
- log.error("Topology is inconsistent...failed to execute member started event");
+ log.error("Member has not initialized. Failed to execute member started event");
}
return;
}
- String clusterId = memberStartedEvent.getClusterId();
- Topology topology = TopologyManager.getTopology();
- Service service = topology.getService(memberStartedEvent.getServiceName());
- Cluster cluster = service.getCluster(memberStartedEvent.getClusterId());
- Member startedMember = cluster.getMember(memberStartedEvent.getMemberId());
- String lbClusterId = cluster.getMember(memberStartedEvent.getMemberId()).getLbClusterId();
-
- // check whether new member is in the same member cluster or LB cluster of this instance
- if (ExtensionUtils.isRelevantMemberEvent(memberStartedEvent.getServiceName(),
- memberStartedEvent.getClusterId(), lbClusterId)) {
- Collection<Member> members = cluster.getMembers();
- Map<String, String> env = new HashMap<String, String>();
- env.put("STRATOS_MEMBER_STARTED_MEMBER_IP", startedMember.getDefaultPrivateIP());
- env.put("STRATOS_MEMBER_STARTED_MEMBER_ID", memberStartedEvent.getMemberId());
- env.put("STRATOS_MEMBER_STARTED_CLUSTER_ID", memberStartedEvent.getClusterId());
- env.put("STRATOS_MEMBER_STARTED_LB_CLUSTER_ID", lbClusterId);
- env.put("STRATOS_MEMBER_STARTED_NETWORK_PARTITION_ID", memberStartedEvent.getNetworkPartitionId());
- env.put("STRATOS_MEMBER_STARTED_SERVICE_NAME", memberStartedEvent.getServiceName());
- env.put("STRATOS_MEMBER_STARTED_MEMBER_LIST_JSON", gson.toJson(members, memberType));
- env.put("STRATOS_MEMBER_STARTED_TOPOLOGY_JSON", gson.toJson(topology.getServices(), serviceType));
- String[] memberIps = ExtensionUtils.getLbMemberIp(lbClusterId);
- if (memberIps != null && memberIps.length > 1) {
- env.put("STRATOS_MEMBER_STARTED_LB_IP", memberIps[0]);
- env.put("STRATOS_MEMBER_STARTED_LB_PUBLIC_IP", memberIps[1]);
- }
- ExtensionUtils.addProperties(service.getProperties(), env, "MEMBER_STARTED_SERVICE_PROPERTY");
- ExtensionUtils.addProperties(cluster.getProperties(), env, "MEMBER_STARTED_CLUSTER_PROPERTY");
- ExtensionUtils.addProperties(startedMember.getProperties(), env, "MEMBER_STARTED_MEMBER_PROPERTY");
- ExtensionUtils.executeMemberStartedExtension(env);
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Member started event is not relevant...skipping agent extension");
- }
- }
- }
-
- private boolean isWKMemberGroupReady(Map<String, String> envParameters, int minCount) {
- Topology topology = TopologyManager.getTopology();
- if (topology == null || !topology.isInitialized()) {
- return false;
- }
- String serviceGroupInPayload = CartridgeAgentConfiguration.getInstance().getServiceGroup();
- if (serviceGroupInPayload != null) {
- envParameters.put("STRATOS_SERVICE_GROUP", serviceGroupInPayload);
- }
-
- // clustering logic for apimanager
- if (serviceGroupInPayload != null && serviceGroupInPayload.equals("apim")) {
-
- // handle apistore and publisher case
- if (CartridgeAgentConfiguration.getInstance().getServiceName().equals("apistore") ||
- CartridgeAgentConfiguration.getInstance().getServiceName().equals("publisher")) {
-
- Collection<Cluster> apistoreClusterCollection = topology.getService("apistore").getClusters();
- Collection<Cluster> publisherClusterCollection = topology.getService("publisher").getClusters();
-
- List<Member> apistoreMemberList = new ArrayList<Member>();
- for (Member member : apistoreClusterCollection.iterator().next().getMembers()) {
- if (member.getStatus().equals(MemberStatus.Starting) || member.getStatus().equals(MemberStatus.Active)) {
- apistoreMemberList.add(member);
- this.wkMembers.add(member);
- }
- }
- if (apistoreMemberList.isEmpty()) {
- if (log.isDebugEnabled()) {
- log.debug("API Store members not yet created");
- }
- return false;
- }
- Member apistoreMember = apistoreMemberList.get(0);
- envParameters.put("STRATOS_WK_APISTORE_MEMBER_IP", apistoreMember.getDefaultPrivateIP());
- if (log.isDebugEnabled()) {
- log.debug("STRATOS_WK_APISTORE_MEMBER_IP: " + apistoreMember.getDefaultPrivateIP());
- }
-
- List<Member> publisherMemberList = new ArrayList<Member>();
- for (Member member : publisherClusterCollection.iterator().next().getMembers()) {
- if (member.getStatus().equals(MemberStatus.Starting) || member.getStatus().equals(MemberStatus.Active)) {
- publisherMemberList.add(member);
- this.wkMembers.add(member);
- }
- }
- if (publisherMemberList.isEmpty()) {
- if (log.isDebugEnabled()) {
- log.debug("API Publisher members not yet created");
- }
- return false;
- }
- Member publisherMember = publisherMemberList.get(0);
- envParameters.put("STRATOS_WK_PUBLISHER_MEMBER_IP", publisherMember.getDefaultPrivateIP());
- if (log.isDebugEnabled()) {
- log.debug("STRATOS_WK_PUBLISHER_MEMBER_IP: " + publisherMember.getDefaultPrivateIP());
- }
-
- return true;
-
- } else if (CartridgeAgentConfiguration.getInstance().getServiceName().equals("gatewaymgt") ||
- CartridgeAgentConfiguration.getInstance().getServiceName().equals("gateway")) {
-
- if (CartridgeAgentConfiguration.getInstance().getDeployment() != null) {
- // check if deployment is Manager Worker separated
- if (CartridgeAgentConfiguration.getInstance().getDeployment().equalsIgnoreCase(CartridgeAgentConstants.DEPLOYMENT_MANAGER) ||
- CartridgeAgentConfiguration.getInstance().getDeployment().equalsIgnoreCase(CartridgeAgentConstants.DEPLOYMENT_WORKER)) {
-
- log.info("Deployment pattern for the node: " + CartridgeAgentConfiguration.getInstance().getDeployment());
- envParameters.put("DEPLOYMENT", CartridgeAgentConfiguration.getInstance().getDeployment());
- // check if WKA members of Manager Worker separated deployment is ready
- return isManagerWorkerWKAGroupReady(envParameters);
- }
- }
-
-
- } else if (CartridgeAgentConfiguration.getInstance().getServiceName().equals("keymanager")) {
- return true;
- }
- } else {
-
- if (CartridgeAgentConfiguration.getInstance().getDeployment() != null) {
- // check if deployment is Manager Worker separated
- if (CartridgeAgentConfiguration.getInstance().getDeployment().equalsIgnoreCase(CartridgeAgentConstants.DEPLOYMENT_MANAGER) ||
- CartridgeAgentConfiguration.getInstance().getDeployment().equalsIgnoreCase(CartridgeAgentConstants.DEPLOYMENT_WORKER)) {
- log.info("Deployment pattern for the node: " + CartridgeAgentConfiguration.getInstance().getDeployment());
- envParameters.put("DEPLOYMENT", CartridgeAgentConfiguration.getInstance().getDeployment());
- // check if WKA members of Manager Worker separated deployment is ready
- return isManagerWorkerWKAGroupReady(envParameters);
- }
- }
-
- String serviceNameInPayload = CartridgeAgentConfiguration.getInstance().getServiceName();
- String clusterIdInPayload = CartridgeAgentConfiguration.getInstance().getClusterId();
- Service service = topology.getService(serviceNameInPayload);
- Cluster cluster = service.getCluster(clusterIdInPayload);
-
- List<Member> wkMembers = new ArrayList<Member>();
- for (Member member : cluster.getMembers()) {
- if (member.getProperties() != null &&
- member.getProperties().containsKey("PRIMARY") &&
- member.getProperties().getProperty("PRIMARY").toLowerCase().equals("true") &&
- (member.getStatus().equals(MemberStatus.Starting) || member.getStatus().equals(MemberStatus.Active))
- ) {
- wkMembers.add(member);
- this.wkMembers.add(member);
- if (log.isDebugEnabled()) {
- log.debug("Found WKA: STRATOS_WK_MEMBER_IP: " + member.getDefaultPrivateIP());
- }
- }
- }
- if (wkMembers.size() >= minCount) {
- int idx = 0;
- for (Member member : wkMembers) {
- envParameters.put("STRATOS_WK_MEMBER_" + idx + "_IP", member.getDefaultPrivateIP());
- if (log.isDebugEnabled()) {
- log.debug("STRATOS_WK_MEMBER_" + idx + "_IP: " + member.getDefaultPrivateIP());
- }
- idx++;
- }
- return true;
- }
- }
- return false;
- }
-
- // generic worker manager separated clustering logic
- private boolean isManagerWorkerWKAGroupReady(Map<String, String> envParameters) {
-
- // for this, we need both manager cluster service name and worker cluster service name
- String managerServiceName = CartridgeAgentConfiguration.getInstance().getManagerServiceName();
- String workerServiceName = CartridgeAgentConfiguration.getInstance().getWorkerServiceName();
-
- // managerServiceName and workerServiceName both should not be null /empty
- if (managerServiceName == null || managerServiceName.isEmpty()) {
- log.error("Manager service name [ " + managerServiceName + " ] is invalid");
- return false;
- }
- if (workerServiceName == null || workerServiceName.isEmpty()) {
- log.error("Worker service name [ " + workerServiceName + " ] is invalid");
- return false;
- }
-
- boolean minManagerInstancesAvailable = false;
- boolean minWorkerInstancesAvailable = false;
-
- TopologyManager.acquireReadLock();
-
- try {
- Service managerService = TopologyManager.getTopology().getService(managerServiceName);
- Service workerService = TopologyManager.getTopology().getService(workerServiceName);
-
- if (managerService == null) {
- log.warn("Service [ " + managerServiceName + " ] is not found");
- return false;
- }
-
- if (workerService == null) {
- log.warn("Service [ " + workerServiceName + " ] is not found");
- return false;
- }
-
- // manager clusters
- Collection<Cluster> managerClusters = managerService.getClusters();
- if (managerClusters == null || managerClusters.isEmpty()) {
- log.warn("No clusters found for service [ " + managerServiceName + " ]");
- return false;
- }
-
- int managerMinInstanceCount = 1;
- boolean managerMinInstanceCountFound = false;
-
- List<Member> managerWkaMembers = new ArrayList<Member>();
- for (Member member : managerClusters.iterator().next().getMembers()) {
-
- if (member.getProperties() != null && member.getProperties().containsKey("PRIMARY") &&
- member.getProperties().getProperty("PRIMARY").toLowerCase().equals("true") &&
- (member.getStatus().equals(MemberStatus.Starting) || member.getStatus().equals(MemberStatus.Active))) {
-
- managerWkaMembers.add(member);
- this.wkMembers.add(member);
-
- // get the min instance count
- if (!managerMinInstanceCountFound) {
- managerMinInstanceCount = getMinInstanceCountFromMemberProperties(member);
- managerMinInstanceCountFound = true;
- log.info("Manager min instance count: " + managerMinInstanceCount);
-
- }
- }
- }
-
- if (managerWkaMembers.size() >= managerMinInstanceCount) {
- minManagerInstancesAvailable = true;
- int idx = 0;
- for (Member member : managerWkaMembers) {
- envParameters.put("STRATOS_WK_MANAGER_MEMBER_" + idx + "_IP", member.getDefaultPrivateIP());
- if (log.isDebugEnabled()) {
- log.debug("STRATOS_WK_MANAGER_MEMBER_" + idx + "_IP: " + member.getDefaultPrivateIP());
- }
- idx++;
- }
-
- envParameters.put("STRATOS_WK_MANAGER_MEMBER_COUNT", Integer.toString(managerMinInstanceCount));
- }
-
- // If all the manager members are non primary and is greate than or equal to mincount,
- // minManagerInstancesAvailable should be true
- boolean allManagersNonPrimary = true;
- for (Member member : managerClusters.iterator().next().getMembers()) {
-
- // get the min instance count
- if (!managerMinInstanceCountFound) {
- managerMinInstanceCount = getMinInstanceCountFromMemberProperties(member);
- managerMinInstanceCountFound = true;
- log.info("Manager min instance count when allManagersNonPrimary true : " + managerMinInstanceCount);
- }
-
- if (member.getProperties() != null && member.getProperties().containsKey("PRIMARY") &&
- member.getProperties().getProperty("PRIMARY").toLowerCase().equals("true")) {
- allManagersNonPrimary = false;
- break;
- }
- }
- if (log.isDebugEnabled()) {
- log.debug(" allManagerNonPrimary & managerMinInstanceCount ["
- + allManagersNonPrimary + "], [" + managerMinInstanceCount + "] ");
- }
- if (allManagersNonPrimary && managerClusters.size() >= managerMinInstanceCount) {
- minManagerInstancesAvailable = true;
- }
-
- // worker cluster
- Collection<Cluster> workerClusters = workerService.getClusters();
- if (workerClusters == null || workerClusters.isEmpty()) {
- log.warn("No clusters found for service [ " + workerServiceName + " ]");
- return false;
- }
-
- int workerMinInstanceCount = 1;
- boolean workerMinInstanceCountFound = false;
-
- List<Member> workerWkaMembers = new ArrayList<Member>();
- for (Member member : workerClusters.iterator().next().getMembers()) {
- if (log.isDebugEnabled()) {
- log.debug("Checking member : " + member.getMemberId());
- }
- if (member.getProperties() != null &&
- member.getProperties().containsKey("PRIMARY") &&
- member.getProperties().getProperty("PRIMARY").toLowerCase().equals("true") &&
- (member.getStatus().equals(MemberStatus.Starting) || member.getStatus().equals(MemberStatus.Active))) {
- if (log.isDebugEnabled()) {
- log.debug("Added worker member " + member.getMemberId());
- }
- workerWkaMembers.add(member);
- this.wkMembers.add(member);
-
- // get the min instance count
- if (!workerMinInstanceCountFound) {
- workerMinInstanceCount = getMinInstanceCountFromMemberProperties(member);
- workerMinInstanceCountFound = true;
- if (log.isDebugEnabled()) {
- log.debug("Worker min instance count: " + workerMinInstanceCount);
- }
- }
- }
- }
-
- if (workerWkaMembers.size() >= workerMinInstanceCount) {
- minWorkerInstancesAvailable = true;
- int idx = 0;
- for (Member member : workerWkaMembers) {
- envParameters.put("STRATOS_WK_WORKER_MEMBER_" + idx + "_IP", member.getDefaultPrivateIP());
- if (log.isDebugEnabled()) {
- log.debug("STRATOS_WK_WORKER_MEMBER_" + idx + "_IP: " + member.getDefaultPrivateIP());
- }
- idx++;
- }
-
- envParameters.put("STRATOS_WK_WORKER_MEMBER_COUNT", Integer.toString(workerMinInstanceCount));
- }
-
- } finally {
- TopologyManager.releaseReadLock();
- }
-
- if (log.isDebugEnabled()) {
- log.debug(" Returnning values minManagerInstancesAvailable && minWorkerInstancesAvailable [" +
- minManagerInstancesAvailable + "], [" + minWorkerInstancesAvailable + "] ");
- }
- return (minManagerInstancesAvailable && minWorkerInstancesAvailable);
- }
-
-
- private int getMinInstanceCountFromMemberProperties(Member member) {
-
- // default value is 1
- int minInstanceCount = 1;
-
- if (member.getProperties().containsKey("MIN_COUNT")) {
- minInstanceCount = Integer.parseInt(member.getProperties().getProperty("MIN_COUNT"));
- }
-
- return minInstanceCount;
- }
-
- private void waitForWKMembers(Map<String, String> envParameters) {
- int minCount = Integer.parseInt(CartridgeAgentConfiguration.getInstance().getMinCount());
- boolean isWKMemberGroupReady = false;
- while (!isWKMemberGroupReady) {
- if (log.isInfoEnabled()) {
- log.info(String.format("Waiting for %d well known members...", minCount));
- }
- try {
- Thread.sleep(5000);
- } catch (InterruptedException e) {
- }
-
- TopologyManager.acquireReadLock();
- isWKMemberGroupReady = isWKMemberGroupReady(envParameters, minCount);
- TopologyManager.releaseReadLock();
- }
+ ExtensionUtils.executeMemberStartedExtension(Collections.<String, String>emptyMap());
}
@Override
@@ -921,51 +415,16 @@ public class DefaultExtensionHandler implements ExtensionHandler {
String clusterIdInPayload = CartridgeAgentConfiguration.getInstance().getClusterId();
String memberIdInPayload = CartridgeAgentConfiguration.getInstance().getMemberId();
- try {
- TopologyManager.acquireReadLock();
- boolean isConsistent = ExtensionUtils.checkTopologyConsistency(serviceNameInPayload,
- clusterIdInPayload, memberIdInPayload);
- if (!isConsistent) {
- if (log.isErrorEnabled()) {
- log.error("Topology is inconsistent...failed to execute start server event");
- }
- return;
- }
- Topology topology = TopologyManager.getTopology();
- Service service = topology.getService(serviceNameInPayload);
- Cluster cluster = service.getCluster(clusterIdInPayload);
-
- // store environment variable parameters to be passed to extension shell script
- Map<String, String> env = new HashMap<String, String>();
-
- // if clustering is enabled wait until all well known members have started
- String flagClustering = CartridgeAgentConfiguration.getInstance().getIsClustered();
- if (flagClustering != null && flagClustering.toLowerCase().equals("true")) {
- env.put("STRATOS_CLUSTERING", "true");
- env.put("STRATOS_WK_MEMBER_COUNT", CartridgeAgentConfiguration.getInstance().getMinCount());
- if (CartridgeAgentConfiguration.getInstance().getIsPrimary().toLowerCase().equals("true")) {
- env.put("STRATOS_PRIMARY", "true");
- } else {
- env.put("STRATOS_PRIMARY", "false");
- }
- TopologyManager.releaseReadLock();
- waitForWKMembers(env);
- if (log.isInfoEnabled()) {
- log.info(String.format("All well known members have started! Resuming start server extension..."));
- }
- TopologyManager.acquireReadLock();
- }
-
- env.put("STRATOS_TOPOLOGY_JSON", gson.toJson(topology.getServices(), serviceType));
- env.put("STRATOS_MEMBER_LIST_JSON", gson.toJson(cluster.getMembers(), memberType));
- ExtensionUtils.executeStartServersExtension(env);
- } catch (Exception e) {
+ boolean isConsistent = ExtensionUtils.checkTopologyConsistency(serviceNameInPayload,
+ clusterIdInPayload, memberIdInPayload);
+ if (!isConsistent) {
if (log.isErrorEnabled()) {
- log.error("Error processing start servers event", e);
+ log.error("Topology is inconsistent...Failed to execute start server event");
}
- } finally {
- TopologyManager.releaseReadLock();
+ return;
}
+
+ ExtensionUtils.executeStartServersExtension(Collections.<String, String>emptyMap());
}
@Override
@@ -1087,17 +546,6 @@ public class DefaultExtensionHandler implements ExtensionHandler {
//ApplicationSignUpRemovedEvent
@Override
public void onApplicationSignUpRemovedEvent(ApplicationSignUpRemovedEvent applicationSignUpRemovedEvent) {
- /*
- * self.log.info(
- "Tenant unsubscribed event received: [tenant] " + application_signup_removal_event.tenantId +
- " [application ID] " + application_signup_removal_event.applicationId
- )
-
- if self.cartridge_agent_config.service_name == application_signup_removal_event.applicationId:
- agentgithandler.AgentGitHandler.remove_repo(application_signup_removal_event.tenant_id)
-
- extensionutils.execute_application_signup_removal_extension({})
- */
if (log.isInfoEnabled()) {
log.info(String.format("applicationSignUpRemovedEvent event received: [appId] %s [tenantId] %s ",
applicationSignUpRemovedEvent.getApplicationId(), applicationSignUpRemovedEvent.getTenantId()));
http://git-wip-us.apache.org/repos/asf/stratos/blob/1a8e7a64/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/ExtensionUtils.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/ExtensionUtils.java b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/ExtensionUtils.java
index 79ac604..2164f9c 100644
--- a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/ExtensionUtils.java
+++ b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/ExtensionUtils.java
@@ -24,10 +24,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.cartridge.agent.config.CartridgeAgentConfiguration;
import org.apache.stratos.common.util.CommandUtils;
-import org.apache.stratos.messaging.domain.topology.Cluster;
-import org.apache.stratos.messaging.domain.topology.Member;
-import org.apache.stratos.messaging.domain.topology.Service;
-import org.apache.stratos.messaging.domain.topology.Topology;
+import org.apache.stratos.messaging.domain.topology.*;
import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
import java.io.File;
@@ -135,66 +132,6 @@ public class ExtensionUtils {
return null;
}
- public static boolean isRelevantMemberEvent(String serviceName, String clusterId, String lbClusterId) {
- String clusterIdInPayload = CartridgeAgentConfiguration.getInstance().getClusterId();
- if (clusterIdInPayload == null) {
- return false;
- }
- Topology topology = TopologyManager.getTopology();
- if (topology == null || !topology.isInitialized()) {
- return false;
- }
-
- if (clusterIdInPayload.equals(clusterId)) {
- return true;
- }
-
- if (clusterIdInPayload.equals(lbClusterId)) {
- return true;
- }
-
- String serviceGroupInPayload = CartridgeAgentConfiguration.getInstance().getServiceGroup();
- if (serviceGroupInPayload != null) {
- Properties serviceProperties = topology.getService(serviceName).getProperties();
- if (serviceProperties == null) {
- return false;
- }
- String memberServiceGroup = serviceProperties.getProperty(CartridgeAgentConstants.SERVICE_GROUP_TOPOLOGY_KEY);
- if (memberServiceGroup != null && memberServiceGroup.equals(serviceGroupInPayload)) {
- if(serviceName.equals(CartridgeAgentConfiguration.getInstance().getServiceName())) {
- if (log.isDebugEnabled()) {
- log.debug("Service names are same");
- }
- return true;
- }else if(CartridgeAgentConfiguration.getInstance().getServiceName().equals("apistore") && "publisher".equals(serviceName)) {
- if (log.isDebugEnabled()) {
- log.debug("Service name in payload is [store]. Serivce name in event is ["+serviceName+"] ");
- }
- return true;
- }else if(CartridgeAgentConfiguration.getInstance().getServiceName().equals("publisher") && "apistore".equals(serviceName)) {
- if (log.isDebugEnabled()) {
- log.debug("Service name in payload is [publisher]. Serivce name in event is ["+serviceName+"] ");
- }
- return true;
- }else if(CartridgeAgentConstants.DEPLOYMENT_WORKER.equals(CartridgeAgentConfiguration.getInstance().getDeployment()) &&
- serviceName.equals(CartridgeAgentConfiguration.getInstance().getManagerServiceName())) {
- if (log.isDebugEnabled()) {
- log.debug("Deployment is worker. Worker's manager service name & service name in event are same");
- }
- return true;
- }else if (CartridgeAgentConstants.DEPLOYMENT_MANAGER.equals(CartridgeAgentConfiguration.getInstance().getDeployment()) &&
- serviceName.equals(CartridgeAgentConfiguration.getInstance().getWorkerServiceName())) {
- if (log.isDebugEnabled()) {
- log.debug("Deployment is manager. Manager's worker service name & service name in event are same");
- }
- return true;
- }
- }
- }
-
- return false;
- }
-
private static Map<String, String> cleanProcessParameters(Map<String, String> envParameters) {
Iterator<Map.Entry<String, String>> iter = envParameters.entrySet().iterator();
while (iter.hasNext()) {
@@ -510,7 +447,7 @@ public class ExtensionUtils {
String command = prepareCommand(script);
addPayloadParameters(envParameters);
cleanProcessParameters(envParameters);
- String output = CommandUtils.executeCommand(command, envParameters);
+ CommandUtils.executeCommand(command, envParameters);
} catch (Exception e) {
log.error("Could not execute tenant subscribed extension", e);
}
@@ -525,7 +462,7 @@ public class ExtensionUtils {
String command = prepareCommand(script);
addPayloadParameters(envParameters);
cleanProcessParameters(envParameters);
- String output = CommandUtils.executeCommand(command, envParameters);
+ CommandUtils.executeCommand(command, envParameters);
} catch (Exception e) {
log.error("Could not execute tenant un-subscribed extension", e);
}
@@ -550,7 +487,7 @@ public class ExtensionUtils {
}
}
- public static boolean checkTopologyConsistency(String serviceName, String clusterId, String memberId){
+ public static boolean checkTopologyConsistency(String serviceName, String clusterId, String memberId) {
Topology topology = TopologyManager.getTopology();
Service service = topology.getService(serviceName);
if (service == null) {
@@ -559,6 +496,7 @@ public class ExtensionUtils {
}
return false;
}
+
Cluster cluster = service.getCluster(clusterId);
if (cluster == null) {
if (log.isErrorEnabled()) {
@@ -566,13 +504,49 @@ public class ExtensionUtils {
}
return false;
}
+
Member activatedMember = cluster.getMember(memberId);
if (activatedMember == null) {
if (log.isErrorEnabled()) {
log.error(String.format("Member id not found in topology [member] %s", memberId));
}
return false;
+ } else if (activatedMember.getStatus() != MemberStatus.Initialized) {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Member found in topology, but not in initialized state [member] %s", memberId));
+ }
+ return false;
+ }
+
+ return true;
+ }
+
+ public static boolean memberExistsInTopology(String serviceNameInPayload, String clusterIdInPayload, String memberIdInPayload) {
+ Topology topology = TopologyManager.getTopology();
+ Service service = topology.getService(serviceNameInPayload);
+ if (service == null) {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Service not found in topology [service] %s", serviceNameInPayload));
+ }
+ return false;
}
+
+ Cluster cluster = service.getCluster(clusterIdInPayload);
+ if (cluster == null) {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Cluster id not found in topology [cluster] %s", clusterIdInPayload));
+ }
+ return false;
+ }
+
+ Member activatedMember = cluster.getMember(memberIdInPayload);
+ if (activatedMember == null) {
+ if (log.isErrorEnabled()) {
+ log.error(String.format("Member id not found in topology [member] %s", memberIdInPayload));
+ }
+ return false;
+ }
+
return true;
}
}