Bringing back zk path creation at launch experiment of Orchestrator to support Gfac transition
@@ -80,6 +80,8 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface {
private final Subscriber statusSubscribe;
private final Subscriber experimentSubscriber;
+ private CuratorFramework curatorClient;
+
/**
* Query orchestrator server to fetch the CPI version
*/
@@ -100,6 +102,7 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface {
orchestrator.getOrchestratorContext().setPublisher(this.publisher);
statusSubscribe = getStatusSubscriber();
experimentSubscriber = getExperimentSubscriber();
+ startCurator();
} catch (OrchestratorException | AiravataException e) {
log.error(e.getMessage(), e);
throw new OrchestratorException("Error while initializing orchestrator service", e);
@@ -132,6 +135,11 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface {
ExperimentModel experiment = null;
final RegistryService.Client registryClient = getRegistryServiceClient();
try {
+ // TODO deprecate this approach as we are replacing gfac
+ String experimentNodePath = getExperimentNodePath (experimentId);
+ ZKPaths.mkdirs(curatorClient.getZookeeperClient().getZooKeeper(), experimentNodePath);
+ String experimentCancelNode = ZKPaths.makePath(experimentNodePath, ZkConstants.ZOOKEEPER_CANCEL_LISTENER_NODE);
+ ZKPaths.mkdirs(curatorClient.getZookeeperClient().getZooKeeper(), experimentCancelNode);
experiment = registryClient.getExperiment(experimentId);
if (experiment == null) {
log.error("Error retrieving the Experiment by the given experimentID: {} ", experimentId);
@@ -441,12 +449,19 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface {
}
orchestrator.cancelExperiment(experimentModel, token);
-
+ // TODO deprecate this approach as we are replacing gfac
+ String expCancelNodePath = ZKPaths.makePath(ZKPaths.makePath(ZkConstants.ZOOKEEPER_EXPERIMENT_NODE,
+ experimentId), ZkConstants.ZOOKEEPER_CANCEL_LISTENER_NODE);
+ Stat stat = curatorClient.checkExists().forPath(expCancelNodePath);
+ if (stat != null) {
+ curatorClient.setData().withVersion(-1).forPath(expCancelNodePath, ZkConstants.ZOOKEEPER_CANCEL_REQEUST
+ .getBytes());
ExperimentStatus status = new ExperimentStatus(ExperimentState.CANCELING);
status.setReason("Experiment cancel request processed");
status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
OrchestratorUtils.updateAndPublishExperimentStatus(experimentId, status, publisher, gatewayId);
log.info("expId : " + experimentId + " :- Experiment status updated to " + status.getState());
+ }
return true;
}
}
@@ -705,4 +720,14 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface {
}
}
+ private void startCurator() throws ApplicationSettingsException {
+ String connectionSting = ServerSettings.getZookeeperConnection();
+ RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
+ curatorClient = CuratorFrameworkFactory.newClient(connectionSting, retryPolicy);
+ curatorClient.start();
+ }
+
+ public String getExperimentNodePath(String experimentId) {
+ return ZKPaths.makePath(ZkConstants.ZOOKEEPER_EXPERIMENT_NODE, experimentId);
+ }
}
