airavata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From shame...@apache.org
Subject [1/4] airavata git commit: Remove Zookeeper code and dependency from Orchestrator.
Date Thu, 28 May 2015 15:37:38 GMT
Repository: airavata
Updated Branches:
  refs/heads/master 5002df55a -> 4a0617802


Remove Zookeeper code and dependency from Orchestrator.


Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/915ed04d
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/915ed04d
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/915ed04d

Branch: refs/heads/master
Commit: 915ed04d54c61e083ef0f2b7dccb7083ef3fee9e
Parents: 5002df5
Author: Shameera Rathanyaka <shameerainfo@gmail.com>
Authored: Wed May 27 16:48:49 2015 -0400
Committer: Shameera Rathanyaka <shameerainfo@gmail.com>
Committed: Wed May 27 16:48:49 2015 -0400

----------------------------------------------------------------------
 .../server/OrchestratorServerHandler.java       | 149 +------------------
 modules/orchestrator/orchestrator-core/pom.xml  |   6 -
 .../core/context/OrchestratorContext.java       |  13 --
 .../core/impl/GFACPassiveJobSubmitter.java      |  25 ----
 4 files changed, 3 insertions(+), 190 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/airavata/blob/915ed04d/modules/orchestrator/airavata-orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java
----------------------------------------------------------------------
diff --git a/modules/orchestrator/airavata-orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java
b/modules/orchestrator/airavata-orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java
index 1b3e1f3..4ef9dbc 100644
--- a/modules/orchestrator/airavata-orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java
+++ b/modules/orchestrator/airavata-orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java
@@ -76,40 +76,21 @@ import org.apache.airavata.registry.cpi.utils.Constants.FieldConstants.WorkflowN
 import org.apache.airavata.workflow.core.WorkflowEnactmentService;
 import org.apache.thrift.TBase;
 import org.apache.thrift.TException;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.Stat;
-
-import java.io.File;
-import java.io.IOException;
+
 import java.util.Arrays;
 import java.util.Calendar;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Random;
-
-public class OrchestratorServerHandler implements OrchestratorService.Iface,
-		Watcher {
-	private static AiravataLogger log = AiravataLoggerFactory
-			.getLogger(OrchestratorServerHandler.class);
 
+public class OrchestratorServerHandler implements OrchestratorService.Iface {
+	private static AiravataLogger log = AiravataLoggerFactory .getLogger(OrchestratorServerHandler.class);
 	private SimpleOrchestratorImpl orchestrator = null;
-
 	private Registry registry;
-
-	private ZooKeeper zk;
-
 	private static Integer mutex = new Integer(-1);
-
 	private String airavataUserName;
 	private String gatewayName;
 	private Publisher publisher;
-
     private RabbitMQProcessConsumer rabbitMQProcessConsumer;
     private RabbitMQProcessPublisher rabbitMQProcessPublisher;
 
@@ -133,31 +114,6 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface,
 			
 //            setGatewayName(ServerSettings.getDefaultUserGateway());
             setAiravataUserName(ServerSettings.getDefaultUser());
-			if(!ServerSettings.isGFacPassiveMode()) {
-				try {
-					zk = new ZooKeeper(zkhostPort, AiravataZKUtils.getZKTimeout(), this); // no watcher
is
-					// required, this
-					// will only use to
-					// store some data
-					log.info("Waiting for zookeeper to connect to the server");
-
-					String OrchServer = ServerSettings
-							.getSetting(org.apache.airavata.common.utils.Constants.ZOOKEEPER_ORCHESTRATOR_SERVER_NODE);
-					synchronized (mutex) {
-						mutex.wait(5000); // waiting for the syncConnected event
-					}
-					registerOrchestratorService(airavataServerHostPort, OrchServer);
-					// creating a watch in orchestrator to monitor the gfac
-					// instances
-					zk.getChildren(ServerSettings.getSetting(
-									Constants.ZOOKEEPER_GFAC_SERVER_NODE, "/gfac-server"),
-							this);
-					log.info("Finished starting ZK: " + zk);
-				} catch (IOException|InterruptedException|KeeperException e) {
-					log.error(e.getMessage(), e);
-					throw new OrchestratorException("Error while initializing orchestrator service, Error
in Zookeeper", e);
-				}
-			}
 		} catch (AiravataException e) {
             log.error(e.getMessage(), e);
             throw new OrchestratorException("Error while initializing orchestrator service",
e);
@@ -169,7 +125,6 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface,
 			orchestrator = new SimpleOrchestratorImpl();
 			registry = RegistryFactory.getDefaultRegistry();
 			orchestrator.initialize();
-			orchestrator.getOrchestratorContext().setZk(this.zk);
 			orchestrator.getOrchestratorContext().setPublisher(this.publisher);
             startProcessConsumer();
         } catch (OrchestratorException e) {
@@ -194,23 +149,6 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface,
 
     }
 
-    private void registerOrchestratorService(String airavataServerHostPort, String orchServer)
throws KeeperException, InterruptedException {
-        Stat zkStat = zk.exists(orchServer, false);
-        if (zkStat == null) {
-            zk.create(orchServer, new byte[0],
-                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-        }
-        String instantNode = orchServer
-                + File.separator
-                + String.valueOf(new Random()
-                        .nextInt(Integer.MAX_VALUE));
-        zkStat = zk.exists(instantNode, false);
-        if (zkStat == null) {
-            zk.create(instantNode, airavataServerHostPort.getBytes(),
-                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
-        }
-    }
-
     /**
 	 * * After creating the experiment Data user have the * experimentID as the
 	 * handler to the experiment, during the launchExperiment * We just have to
@@ -327,84 +265,6 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface,
         return validateStatesAndCancel(experimentId, tokenId);
 	}
 
-	/**
-	 * This method gracefully handler gfac node failures
-	 */
-	synchronized public void process(WatchedEvent watchedEvent) {
-		log.info(watchedEvent.getPath());
-		synchronized (mutex) {
-			try {
-				Event.KeeperState state = watchedEvent.getState();
-				switch (state) {
-					case SyncConnected:
-						mutex.notify();
-						break;
-					case Expired:
-					case Disconnected:
-						log.info("ZK Connection is "+ state.toString());
-						break;
-				}
-				if (watchedEvent.getPath() != null
-						&& watchedEvent.getPath().startsWith(
-						ServerSettings.getSetting(
-								Constants.ZOOKEEPER_GFAC_SERVER_NODE,
-								"/gfac-server"))) {
-					List<String> children = zk.getChildren(ServerSettings
-							.getSetting(Constants.ZOOKEEPER_GFAC_SERVER_NODE,
-									"/gfac-server"), true);
-					for (String gfacNodes : children) {
-						zk.exists(
-								ServerSettings.getSetting(
-										Constants.ZOOKEEPER_GFAC_SERVER_NODE,
-										"/gfac-server")
-										+ File.separator + gfacNodes, this);
-					}
-					switch (watchedEvent.getType()) {
-						case NodeCreated:
-							mutex.notify();
-							break;
-						case NodeDeleted:
-							// here we have to handle gfac node shutdown case
-							if (children.size() == 0) {
-								log.error("There are not gfac instances to route failed jobs");
-								return;
-							}
-							// we recover one gfac node at a time
-							final WatchedEvent event = watchedEvent;
-							final OrchestratorServerHandler handler = this;
-						/*(new Thread() {  // disabling ft implementation with zk
-							public void run() {
-								int retry = 0;
-								while (retry < 3) {
-									try {
-										(new OrchestratorRecoveryHandler(
-												handler, event.getPath()))
-												.recover();
-										break;
-									} catch (Exception e) {
-										e.printStackTrace();
-										log.error("error recovering the jobs for gfac-node: "
-												+ event.getPath());
-										log.error("Retrying again to recover jobs and retry attempt: "
-												+ ++retry);
-									}
-								}
-
-							}
-						}).start();*/
-							break;
-					}
-
-
-				}
-			} catch (KeeperException e) {
-				e.printStackTrace();
-			} catch (InterruptedException e) {
-				e.printStackTrace();
-			}
-		}
-	}
-
 	private String getAiravataUserName() {
 		return airavataUserName;
 	}
@@ -513,9 +373,6 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface,
         try {
             Experiment experiment = (Experiment) registry.get(
                     RegistryModelType.EXPERIMENT, experimentId);
-            if (zk == null || !zk.getState().isConnected()){
-                zk = new ZooKeeper(AiravataZKUtils.getZKhostPort(), AiravataZKUtils.getZKTimeout(),this);
-            }
 			log.info("Waiting for zookeeper to connect to the server");
 			synchronized (mutex){
 				mutex.wait(5000);

http://git-wip-us.apache.org/repos/asf/airavata/blob/915ed04d/modules/orchestrator/orchestrator-core/pom.xml
----------------------------------------------------------------------
diff --git a/modules/orchestrator/orchestrator-core/pom.xml b/modules/orchestrator/orchestrator-core/pom.xml
index 6cfc17f..a5c41ba 100644
--- a/modules/orchestrator/orchestrator-core/pom.xml
+++ b/modules/orchestrator/orchestrator-core/pom.xml
@@ -116,12 +116,6 @@ the License. -->
             <artifactId>airavata-server-configuration</artifactId>
             <scope>test</scope>
         </dependency>
-        <!-- zookeeper and curator dependencies -->
-        <dependency>
-        	<groupId>org.apache.zookeeper</groupId>
-        	<artifactId>zookeeper</artifactId>
-        	<version>${zk.version}</version>
-        </dependency>
     </dependencies>
 
 </project>

http://git-wip-us.apache.org/repos/asf/airavata/blob/915ed04d/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/context/OrchestratorContext.java
----------------------------------------------------------------------
diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/context/OrchestratorContext.java
b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/context/OrchestratorContext.java
index b77087b..f7f5969 100644
--- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/context/OrchestratorContext.java
+++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/context/OrchestratorContext.java
@@ -34,13 +34,8 @@ import org.apache.zookeeper.ZooKeeper;
  */
 public class OrchestratorContext {
     private List<GFACInstance> gfacInstanceList;
-
     private OrchestratorConfiguration orchestratorConfiguration;
-
     private Registry newRegistry;
-
-    private static ZooKeeper zk; // this instance can be accessed by the Validators and other
components
-
     private Publisher publisher;
     
     public OrchestratorContext(List<GFACInstance> gfacInstanceList) {
@@ -90,12 +85,4 @@ public class OrchestratorContext {
     public void setGfacInstanceList(List<GFACInstance> gfacInstanceList) {
         this.gfacInstanceList.addAll(gfacInstanceList);
     }
-
-    public  void setZk(ZooKeeper zk) {
-        this.zk = zk;
-    }
-
-    public static ZooKeeper getZk() {
-        return zk;
-    }
 }

http://git-wip-us.apache.org/repos/asf/airavata/blob/915ed04d/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/impl/GFACPassiveJobSubmitter.java
----------------------------------------------------------------------
diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/impl/GFACPassiveJobSubmitter.java
b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/impl/GFACPassiveJobSubmitter.java
index 1faef9e..36282a0 100644
--- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/impl/GFACPassiveJobSubmitter.java
+++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/impl/GFACPassiveJobSubmitter.java
@@ -21,17 +21,11 @@
 package org.apache.airavata.orchestrator.core.impl;
 
 import org.apache.airavata.common.exception.AiravataException;
-import org.apache.airavata.common.exception.ApplicationSettingsException;
 import org.apache.airavata.common.utils.AiravataUtils;
-import org.apache.airavata.common.utils.AiravataZKUtils;
-import org.apache.airavata.common.utils.Constants;
 import org.apache.airavata.common.utils.ServerSettings;
 import org.apache.airavata.credential.store.store.CredentialReader;
 import org.apache.airavata.gfac.client.GFACInstance;
-import org.apache.airavata.gfac.client.GFacClientFactory;
-import org.apache.airavata.gfac.core.context.JobExecutionContext;
 import org.apache.airavata.gfac.core.utils.GFacUtils;
-import org.apache.airavata.gfac.cpi.GfacService;
 import org.apache.airavata.messaging.core.MessageContext;
 import org.apache.airavata.messaging.core.Publisher;
 import org.apache.airavata.messaging.core.PublisherFactory;
@@ -41,17 +35,11 @@ import org.apache.airavata.model.messaging.event.TaskTerminateEvent;
 import org.apache.airavata.orchestrator.core.context.OrchestratorContext;
 import org.apache.airavata.orchestrator.core.exception.OrchestratorException;
 import org.apache.airavata.orchestrator.core.job.JobSubmitter;
-import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooKeeper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
-import java.io.IOException;
-import java.util.List;
-import java.util.Random;
 import java.util.UUID;
 
 /**
@@ -59,13 +47,9 @@ import java.util.UUID;
  */
 public class GFACPassiveJobSubmitter implements JobSubmitter,Watcher {
     private final static Logger logger = LoggerFactory.getLogger(GFACPassiveJobSubmitter.class);
-
     public static final String IP = "ip";
-
     private OrchestratorContext orchestratorContext;
-
     private static Integer mutex = -1;
-
     private Publisher publisher;
 
     public void initialize(OrchestratorContext orchestratorContext) throws OrchestratorException
{
@@ -160,15 +144,6 @@ public class GFACPassiveJobSubmitter implements JobSubmitter,Watcher
{
         }
     }
 
-    private void closeZK(OrchestratorContext orchestratorContext) {
-        try {
-            if(orchestratorContext!=null && orchestratorContext.getZk()!=null) {
-                orchestratorContext.getZk().close();
-            }
-        } catch (InterruptedException e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
     synchronized public void process(WatchedEvent event) {
         logger.info(getClass().getName() + event.getPath());
         logger.info(getClass().getName()+event.getType());


Mime
View raw message