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:40 UTC
[03/11] stratos git commit: JCA - Moved port check to instance
activated event publishing. If ports are not active InstanceActivatedEvent
will not be published. JCA - Code cleanup - CartridgeAgent.java
JCA - Moved port check to instance activated event publishing. If ports are not active InstanceActivatedEvent will not be published.
JCA - Code cleanup - CartridgeAgent.java
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/d27c262b
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/d27c262b
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/d27c262b
Branch: refs/heads/master
Commit: d27c262bde13637259da0487fc1d649e5c7033c2
Parents: d0c3d89
Author: Chamila de Alwis <ch...@wso2.com>
Authored: Fri Feb 20 15:28:33 2015 +0530
Committer: Imesh Gunaratne <im...@apache.org>
Committed: Mon Feb 23 15:35:34 2015 +0530
----------------------------------------------------------------------
.../stratos/cartridge/agent/CartridgeAgent.java | 93 ++++++++------------
.../publisher/CartridgeAgentEventPublisher.java | 81 +++++++++++------
.../agent/util/CartridgeAgentUtils.java | 13 ++-
3 files changed, 101 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/d27c262b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
index 4bb3a66..6a22286 100644
--- a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
+++ b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
@@ -31,7 +31,7 @@ import org.apache.stratos.cartridge.agent.event.publisher.CartridgeAgentEventPub
import org.apache.stratos.cartridge.agent.extensions.DefaultExtensionHandler;
import org.apache.stratos.cartridge.agent.extensions.ExtensionHandler;
import org.apache.stratos.cartridge.agent.util.CartridgeAgentConstants;
-import org.apache.stratos.cartridge.agent.util.CartridgeAgentUtils;
+import org.apache.stratos.cartridge.agent.util.ExtensionUtils;
import java.util.List;
@@ -47,11 +47,6 @@ public class CartridgeAgent implements Runnable {
private CartridgeAgentEventListeners eventListenerns;
- // We have an asynchronous activity running to respond to ADC updates. We want to ensure
- // that no publishInstanceActivatedEvent() call is made *before* the port activation test
- // has succeeded. This flag controls that.
- private boolean portsActivated;
-
@Override
public void run() {
if (log.isInfoEnabled()) {
@@ -61,60 +56,60 @@ public class CartridgeAgent implements Runnable {
eventListenerns = new CartridgeAgentEventListeners();
validateRequiredSystemProperties();
-
-
if (log.isInfoEnabled()) {
log.info("Cartridge agent validated system properties done");
}
- // Start instance notifier listener thread
- portsActivated = false;
- subscribeToTopicsAndRegisterListeners();
-
- if (log.isInfoEnabled()) {
- log.info("Cartridge agent subscribeToTopicsAndRegisterListeners done");
- }
-
// Start topology event receiver thread
registerTopologyEventListeners();
-
if (log.isInfoEnabled()) {
log.info("Cartridge agent registerTopologyEventListeners done");
}
+
+ if (log.isInfoEnabled()){
+ log.info("Waiting for CompleteTopologyEvent..");
+ }
+
+ ExtensionUtils.waitForCompleteTopology();
+ if (log.isInfoEnabled()){
+ log.info("CompleteTopologyEvent received.");
+ }
+
+ // wait till the member spawned event
+ while (!CartridgeAgentConfiguration.getInstance().isInitialized()) {
+ try {
+ if (log.isDebugEnabled()) {
+ log.info("Waiting for Cartridge Agent to be initialized...");
+ }
+ Thread.sleep(1000);
+ } catch (InterruptedException ignore) {
+ }
+ }
+
+ if (log.isInfoEnabled()) {
+ log.info("Cartridge agent initialized done");
+ }
+
+ // Start instance notifier listener thread
+ registerInstanceNotifierEventListeners();
+ if (log.isInfoEnabled()) {
+ log.info("Cartridge agent registerInstanceNotifierEventListeners done");
+ }
// Start tenant event receiver thread
registerTenantEventListeners();
-
if (log.isInfoEnabled()) {
- log.info("Cartridge agent registerApplicationEventListeners done");
+ log.info("Cartridge agent registerTenantEventListeners done");
}
// Start application event receiver thread
registerApplicationEventListeners();
-
-
if (log.isInfoEnabled()) {
log.info("Cartridge agent registering all event listeners ... done");
}
-
- // wait till the member spawned event
- while (!CartridgeAgentConfiguration.getInstance().isInitialized()) {
- try {
- if (log.isDebugEnabled()) {
- log.info("Waiting for Cartridge Agent to be initialized...");
- }
- Thread.sleep(1000);
- } catch (InterruptedException ignore) {
- }
- }
-
- if (log.isInfoEnabled()) {
- log.info("Cartridge agent initialized done");
- }
// Execute instance started shell script
extensionHandler.onInstanceStartedEvent();
-
if (log.isInfoEnabled()) {
log.info("Cartridge agent onInstanceStartedEvent done");
}
@@ -135,45 +130,33 @@ public class CartridgeAgent implements Runnable {
log.info("Cartridge agent startServerExtension done");
}
- // Wait for all ports to be active
- CartridgeAgentUtils.waitUntilPortsActive(CartridgeAgentConfiguration.getInstance().getListenAddress(),
- CartridgeAgentConfiguration.getInstance().getPorts());
- portsActivated = true;
-
- if (log.isInfoEnabled()) {
- log.info("Cartridge agent portsActivated done");
- }
-
// Publish instance activated event
CartridgeAgentEventPublisher.publishInstanceActivatedEvent();
-
if (log.isInfoEnabled()) {
log.info("Cartridge agent publishInstanceActivatedEvent done");
}
// Check repo url
String repoUrl = CartridgeAgentConfiguration.getInstance().getRepoUrl();
-
if (log.isInfoEnabled()) {
log.info("Cartridge agent getRepoUrl done");
}
if ("null".equals(repoUrl) || StringUtils.isBlank(repoUrl)) {
if (log.isInfoEnabled()) {
- log.info("No artifact repository found");
+ log.info("No artifact repository found. Publishing InstanceActivatedEvent.");
}
+
+ // Publish instance activated event
+ CartridgeAgentEventPublisher.publishInstanceActivatedEvent();
// Execute instance activated shell script
extensionHandler.onInstanceActivatedEvent();
-
if (log.isInfoEnabled()) {
log.info("Cartridge agent onInstanceActivatedEvent done");
}
-
- // Publish instance activated event
- CartridgeAgentEventPublisher.publishInstanceActivatedEvent();
} else {
if (log.isInfoEnabled()) {
- log.info("Cartridge agent - artifact repository found");
+ log.info("Artifact repository found. Waiting for ArtifactUpdatedEvent to commence cloning.");
}
}
@@ -197,7 +180,7 @@ public class CartridgeAgent implements Runnable {
logPublisherManager.stop();
}
- protected void subscribeToTopicsAndRegisterListeners() {
+ protected void registerInstanceNotifierEventListeners() {
if (log.isDebugEnabled()) {
log.debug("SsubscribeToTopicsAndRegisterListeners before");
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/d27c262b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/event/publisher/CartridgeAgentEventPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/event/publisher/CartridgeAgentEventPublisher.java b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/event/publisher/CartridgeAgentEventPublisher.java
index 8aa6e78..9a4e0c2 100644
--- a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/event/publisher/CartridgeAgentEventPublisher.java
+++ b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/event/publisher/CartridgeAgentEventPublisher.java
@@ -24,6 +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.cartridge.agent.statistics.publisher.HealthStatisticsNotifier;
+import org.apache.stratos.cartridge.agent.util.CartridgeAgentUtils;
import org.apache.stratos.messaging.broker.publish.EventPublisher;
import org.apache.stratos.messaging.broker.publish.EventPublisherPool;
import org.apache.stratos.messaging.event.instance.status.InstanceActivatedEvent;
@@ -32,6 +33,8 @@ import org.apache.stratos.messaging.event.instance.status.InstanceReadyToShutdow
import org.apache.stratos.messaging.event.instance.status.InstanceStartedEvent;
import org.apache.stratos.messaging.util.MessagingUtil;
+import java.util.List;
+
/**
* Cartridge agent event publisher.
*/
@@ -102,35 +105,55 @@ public class CartridgeAgentEventPublisher {
public static void publishInstanceActivatedEvent() {
if (!isActivated()) {
- if (log.isInfoEnabled()) {
- log.info("Publishing instance activated event");
- }
- InstanceActivatedEvent event = new InstanceActivatedEvent(
- CartridgeAgentConfiguration.getInstance().getServiceName(),
- CartridgeAgentConfiguration.getInstance().getClusterId(),
- CartridgeAgentConfiguration.getInstance().getMemberId(),
- CartridgeAgentConfiguration.getInstance().getInstanceId(),
- CartridgeAgentConfiguration.getInstance().getClusterInstanceId(),
- CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
- CartridgeAgentConfiguration.getInstance().getPartitionId());
-
- // Event publisher connection will
- String topic = MessagingUtil.getMessageTopicName(event);
- EventPublisher eventPublisher = EventPublisherPool
- .getPublisher(topic);
- eventPublisher.publish(event);
- if (log.isInfoEnabled()) {
- log.info("Instance activated event published");
- }
-
- if (log.isInfoEnabled()) {
- log.info("Starting health statistics notifier");
- }
- Thread thread = new Thread(new HealthStatisticsNotifier());
- thread.start();
- setActivated(true);
- if (log.isInfoEnabled()) {
- log.info("Health statistics notifier started");
+ // Wait for all ports to be active, if ports are not activated, do not publish instance activated since
+ // the service is not up
+ List<Integer> ports = CartridgeAgentConfiguration.getInstance().getPorts();
+ String listenAddress = CartridgeAgentConfiguration.getInstance().getListenAddress();
+ boolean portsActivated = CartridgeAgentUtils.waitUntilPortsActive(listenAddress, ports);
+
+ if (portsActivated) {
+ if (log.isInfoEnabled()) {
+ log.info("Publishing instance activated event");
+ }
+ InstanceActivatedEvent event = new InstanceActivatedEvent(
+ CartridgeAgentConfiguration.getInstance().getServiceName(),
+ CartridgeAgentConfiguration.getInstance().getClusterId(),
+ CartridgeAgentConfiguration.getInstance().getMemberId(),
+ CartridgeAgentConfiguration.getInstance().getInstanceId(),
+ CartridgeAgentConfiguration.getInstance().getClusterInstanceId(),
+ CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
+ CartridgeAgentConfiguration.getInstance().getPartitionId());
+
+ // Event publisher connection will
+ String topic = MessagingUtil.getMessageTopicName(event);
+ EventPublisher eventPublisher = EventPublisherPool
+ .getPublisher(topic);
+ eventPublisher.publish(event);
+ if (log.isInfoEnabled()) {
+ log.info("Instance activated event published");
+ }
+
+ if (log.isInfoEnabled()) {
+ log.info("Starting health statistics notifier");
+ }
+ Thread thread = new Thread(new HealthStatisticsNotifier());
+ thread.start();
+ setActivated(true);
+ if (log.isInfoEnabled()) {
+ log.info("Health statistics notifier started");
+ }
+ } else {
+ if (log.isInfoEnabled()) {
+ // There would not be a large number of ports
+ String portsStr = "";
+ for (Integer port: ports){
+ portsStr += port + ", ";
+ }
+ log.info(String.format(
+ "Ports activation timed out. Aborting InstanceActivatedEvent publishing. [IPAdress] %s [Ports] %s",
+ listenAddress,
+ portsStr));
+ }
}
} else {
if (log.isWarnEnabled()) {
http://git-wip-us.apache.org/repos/asf/stratos/blob/d27c262b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/CartridgeAgentUtils.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/CartridgeAgentUtils.java b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/CartridgeAgentUtils.java
index 978ae5f..3dfe50a 100644
--- a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/CartridgeAgentUtils.java
+++ b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/util/CartridgeAgentUtils.java
@@ -79,7 +79,13 @@ public class CartridgeAgentUtils {
return decryptPassword;
}
- public static void waitUntilPortsActive(String ipAddress, List<Integer> ports) {
+ /**
+ * Blocks until the provided ports are activated on the provided ip address or the port check timeout exceeds
+ * @param ipAddress
+ * @param ports
+ * @return true if ports are activated, false if timeout exceeded
+ */
+ public static boolean waitUntilPortsActive(String ipAddress, List<Integer> ports) {
long portCheckTimeOut = 1000 * 60 * 10;
String portCheckTimeOutStr = System.getProperty("port.check.timeout");
if (StringUtils.isNotBlank(portCheckTimeOutStr)) {
@@ -99,16 +105,19 @@ public class CartridgeAgentUtils {
long endTime = System.currentTimeMillis();
long duration = endTime - startTime;
if (duration > portCheckTimeOut) {
- return;
+ return false;
}
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
}
}
+
if(log.isInfoEnabled()) {
log.info("Ports activated: [ip] " + ipAddress + " [ports] "+ports);
}
+
+ return true;
}
public static boolean checkPortsActive(String ipAddress, List<Integer> ports) {