Skip to content

Z
ZooKeeper-Service-Reg-and-Disc-AutoHealer
Commit 08925d8d authored by mohammad.salama's avatar mohammad.salama
Browse files
Options

Third Commit: code to use ssh is implmented , problem so many workers

parent 5d0042c4
Show whitespace changes
Inline Side-by-side
Showing with 80 additions and 57 deletions
pom.xml
View file @ 08925d8d
......@@ -32,5 +32,15 @@
<artifactId>zookeeper</artifactId>
<version>3.9.1</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.3</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.32</version>
</dependency>
</dependencies>
</project>
\ No newline at end of file
src/main/java/Application.java
View file @ 08925d8d
......@@ -17,12 +17,19 @@ public class Application implements Watcher {
public static int numberOfInstances;
public static String pathToFile = "";
private static final String RELATIVE_PATH_TO_JARS = "/out/artifacts/";
/**
* input for jars to execute must be their relative paths
* inside the out/artifact/ directory
* */
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
int currentServerPort = args.length == 3 ? Integer.parseInt(args[0]) : DEFAULT_PORT;
numberOfInstances = args.length == 3 ? Integer.parseInt(args[1]) : 100;
pathToFile = args.length == 3 ? args[2] : "D:\\HIAST\\FIY\\FS\\Distributed Systems\\Lab\\6\\DS-06\\Registration&Discovery-AutoHealer\\out\\artifacts\\TransientWorker_jar\\Registration&Discovery-AutoHealer.jar";
numberOfInstances = args.length == 3 ? Integer.parseInt(args[1]) : 10;
pathToFile = args.length == 3 ? args[2] : "TransientWorker_jar/Registration&Discovery-AutoHealer.jar";
pathToFile = System.getProperty("user.dir") + RELATIVE_PATH_TO_JARS + pathToFile;
Application application = new Application();
ZooKeeper zooKeeper = application.connectToZookeeper();
......
src/main/java/LeaderElection.java
View file @ 08925d8d
......@@ -8,7 +8,7 @@ import java.util.List;
public class LeaderElection implements Watcher {
private static final String AUTOHEALER_ZNODES_PATH = "/physical_nodes";
private static final String PHYSICAL_ZNODES_PATH = "/physical_nodes";
private String currentZnodeName;
private ZooKeeper zooKeeper;
......@@ -20,12 +20,15 @@ public class LeaderElection implements Watcher {
}
public void volunteerForLeadership() throws InterruptedException, KeeperException {
String znodePrefix = AUTOHEALER_ZNODES_PATH + "/node_";
String znodeFullPath = zooKeeper.create(znodePrefix, new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
public void volunteerForLeadership() throws InterruptedException, KeeperException, UnknownHostException {
String znodePrefix = PHYSICAL_ZNODES_PATH + "/node_";
String IP = InetAddress.getLocalHost().getHostAddress();
String username = System.getProperty("user.name")+"@"+IP;
String znodeFullPath = zooKeeper.create(znodePrefix, username.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(znodeFullPath);
this.currentZnodeName = znodeFullPath.replace(AUTOHEALER_ZNODES_PATH + "/", "");
this.currentZnodeName = znodeFullPath.replace(PHYSICAL_ZNODES_PATH + "/", "");
}
public void reelectLeader() throws InterruptedException, KeeperException {
......@@ -35,20 +38,12 @@ public class LeaderElection implements Watcher {
//this while to guarantee get predecessor even if it deleted just before zookeeper.exist
while (predecessorStat == null)
{
List<String> children = zooKeeper.getChildren(AUTOHEALER_ZNODES_PATH, false);
List<String> children = zooKeeper.getChildren(PHYSICAL_ZNODES_PATH, false);
Collections.sort(children);
String smallestChild = children.get(0); //the first element
if (smallestChild.equals(currentZnodeName)) {
System.out.println("I'm a leader");
try
{
String IP = InetAddress.getLocalHost().getHostAddress();
String username = System.getProperty("user.name")+"@"+IP;
zooKeeper.setData(AUTOHEALER_ZNODES_PATH + "/"+currentZnodeName , username.getBytes() , -1);
} catch (UnknownHostException e) {
throw new RuntimeException(e);
}
onElectionCallback.onElectedToBeLeader();
return;
}
......@@ -56,7 +51,7 @@ public class LeaderElection implements Watcher {
System.out.println("I'm not a leader");
int predecessorIndex = children.indexOf(currentZnodeName) - 1;
predecessorName = children.get(predecessorIndex);
predecessorStat = zooKeeper.exists(AUTOHEALER_ZNODES_PATH + "/" + predecessorName, this);
predecessorStat = zooKeeper.exists(PHYSICAL_ZNODES_PATH + "/" + predecessorName, this);
}
}
onElectionCallback.onWorker();
......
src/main/java/ServiceRegistry.java
View file @ 08925d8d
......@@ -10,7 +10,7 @@ import java.util.List;
public class ServiceRegistry implements Watcher
{
private static final String WORKERS_ZNODES_PATH = "/workers";
private static final String AUTOHEALER_ZNODES_PATH = "/physical_nodes";
private static final String PHYSICAL_ZNODES_PATH = "/physical_nodes";
private String pathToProgram = "";
private final ZooKeeper zooKeeper;
......@@ -34,8 +34,8 @@ public class ServiceRegistry implements Watcher
if (zooKeeper.exists(WORKERS_ZNODES_PATH, false) == null) {
zooKeeper.create(WORKERS_ZNODES_PATH, new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
if (zooKeeper.exists(AUTOHEALER_ZNODES_PATH, false) == null) {
zooKeeper.create(AUTOHEALER_ZNODES_PATH, new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
if (zooKeeper.exists(PHYSICAL_ZNODES_PATH, false) == null) {
zooKeeper.create(PHYSICAL_ZNODES_PATH, new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
}
catch (KeeperException | InterruptedException e) {
......@@ -47,11 +47,11 @@ public class ServiceRegistry implements Watcher
System.out.println(metadata);
if (this.currentZnode != null)
{
zooKeeper.setData(AUTOHEALER_ZNODES_PATH + "/"+currentZnode , metadata.getBytes() , -1);
zooKeeper.setData(PHYSICAL_ZNODES_PATH + "/"+currentZnode , metadata.getBytes() , -1);
System.out.println("Already registered to service registry");
return;
}
this.currentZnode = zooKeeper.create(AUTOHEALER_ZNODES_PATH + "/node_", metadata.getBytes(),
this.currentZnode = zooKeeper.create(PHYSICAL_ZNODES_PATH + "/node_", metadata.getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("Registered to service registry");
......@@ -85,14 +85,15 @@ public class ServiceRegistry implements Watcher
updateAddresses();
launchWorkersIfNecessary();
}
private synchronized void updateAddresses() throws KeeperException, InterruptedException
{
List<String> workerZnodes = zooKeeper.getChildren(AUTOHEALER_ZNODES_PATH, this);
List<String> workerZnodes = zooKeeper.getChildren(PHYSICAL_ZNODES_PATH, this);
List<String> addresses = new ArrayList<>(workerZnodes.size());
for (String workerZnode : workerZnodes) {
String workerFullPath = AUTOHEALER_ZNODES_PATH + "/" + workerZnode;
String workerFullPath = PHYSICAL_ZNODES_PATH + "/" + workerZnode;
Stat stat = zooKeeper.exists(workerFullPath, false);
if (stat == null) {
continue;
......@@ -107,12 +108,8 @@ public class ServiceRegistry implements Watcher
System.out.println("The cluster addresses are: " + this.allServiceAddresses);
}
/**
* this method misses the implementation of ssh to use startNewWorker
* */
private void launchWorkersIfNecessary() throws KeeperException, InterruptedException, IOException {
List<String> physicalZnodes = zooKeeper.getChildren(AUTOHEALER_ZNODES_PATH, false);
List<String> physicalZnodes = zooKeeper.getChildren(PHYSICAL_ZNODES_PATH, false);
List<String> workers = zooKeeper.getChildren(WORKERS_ZNODES_PATH , this);
int neededInstances = numberOfInstances - workers.size();
......@@ -125,7 +122,7 @@ public class ServiceRegistry implements Watcher
while (neededInstances > 0 && sortedWorkers.size()>0)
{
Stat stat = zooKeeper.exists(AUTOHEALER_ZNODES_PATH+"/"+sortedWorkers.get(index) ,false);
Stat stat = zooKeeper.exists(PHYSICAL_ZNODES_PATH+"/"+sortedWorkers.get(index) ,false);
if (stat == null)
{
code++;
......@@ -135,13 +132,9 @@ public class ServiceRegistry implements Watcher
}
continue;
}
/*
* ssh to execute pathToProgram in the coherent IP for the node
* */
startNewWorker(sortedWorkers.get(index));
neededInstances--;
index = (index+1)%sortedWorkers.size();
/// System.out.println("Worker " + uuuu);
}
}
......@@ -160,15 +153,31 @@ public class ServiceRegistry implements Watcher
private void startNewWorker(String physicalNode) throws IOException, InterruptedException, KeeperException
{
File file = new File(pathToProgram);
String command = "java -Dorg.slf4j.simpleLogger.defaultLogLevel=off -jar " + file.getName() + " " + physicalNode;
String ip = new String(zooKeeper.getData(AUTOHEALER_ZNODES_PATH+"/"+physicalNode , false, null));
System.out.println("Sendig job to " + ip);
/*
* must execute command over ssh here !!
* */
String remoteUser = new String(zooKeeper.getData(PHYSICAL_ZNODES_PATH+"/"+physicalNode , false, null));
String remoteDirectory = "/JavaJars/";
String remoteJarFilePath = remoteDirectory + file.getName();
String scpCommand = "scp " + pathToProgram + " " + remoteUser + ":" + remoteDirectory;
Runtime.getRuntime().exec(command, null, file.getParentFile());
String sshCommand = "ssh " + remoteUser +
" 'java -Dorg.slf4j.simpleLogger.defaultLogLevel=off -jar " +
remoteJarFilePath + " ' ";
Process scpProcess = Runtime.getRuntime().exec(scpCommand);
scpProcess.waitFor();
if (scpProcess.exitValue() == 0) {
Process sshProcess = Runtime.getRuntime().exec(sshCommand);
}
/**
*String command = "java -Dorg.slf4j.simpleLogger.defaultLogLevel=off -jar " + file.getName() + " " + physicalNode;
* System.out.println("Sendig job to " + ip + " of node : " + physicalNode);
* Runtime.getRuntime().exec(command, null, file.getParentFile());
*/
}
@Override
......
src/main/java/TestLogging.java
View file @ 08925d8d
......@@ -3,10 +3,10 @@ import ch.qos.logback.core.FileAppender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MyClass {
private static final Logger logger = LoggerFactory.getLogger(MyClass.class);
public class TestLogging {
private static final Logger logger = LoggerFactory.getLogger(TestLogging.class);
/*public static void changeLogFile(String newLogFile) {
/* public static void changeLogFile(String newLogFile) {
LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
FileAppender fileAppender = (FileAppender) loggerContext.getLogger("ROOT").getAppender("FILE");
fileAppender.setFile(newLogFile);
......@@ -16,15 +16,19 @@ public class MyClass {
loggerContext.getLogger("ROOT").addAppender(fileAppender);
}*/
public static void someMethod() {
public static void tryLogging() {
logger.debug("Debug log message");
logger.info("Info log message");
logger.error("Error log message");
}
public static void main(String[] args) {
someMethod();
//tryLogging();
//changeLogFile("logs/MyClassLogs.logs");
// someMethod();
String workingDir = System.getProperty("user.dir");
System.out.println("Current working directory: " + workingDir);
/// out put : D:\HIAST\FIY\FS\Distributed Systems\Lab\6\DS-06\Registration&Discovery-AutoHealer
/// out/artifacts/TransientWorker_jar/Registration&Discovery-AutoHealer.jar
}
}
\ No newline at end of file
src/main/java/TransientWorkerApplication.java
View file @ 08925d8d
......@@ -17,7 +17,7 @@ public class TransientWorkerApplication
worker.work(nodeNum);
}
catch (Exception e) {
System.exit(1);
}
}
}
src/main/resources/logback.xml
View file @ 08925d8d
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<Appenders name="FILE" target="ch.qos.logback.core.FileAppender">
<file>logs/mylog.log</file>
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</Console>
</Appenders>
<Loggers>
<Root level="WARN">
<AppenderRef ref="Console"/>
<AppenderRef ref="FILE"/>
</Root>
</Loggers>
</Configuration>
\ No newline at end of file
target/classes/logback.xml
View file @ 08925d8d
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<Appenders name="FILE" target="ch.qos.logback.core.FileAppender">
<file>logs/mylog.log</file>
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</Console>
</Appenders>
<Loggers>
<Root level="WARN">
<AppenderRef ref="Console"/>
<AppenderRef ref="FILE"/>
</Root>
</Loggers>
</Configuration>
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

HIAST - GitLab