Skip to content

R
registration-and-discovery-autohealing-loadbalancing-zookeeper_Homework
Commit e4290b58 authored by Ali Saeed's avatar Ali Saeed
Browse files
Options

auto healing homework

parents
Hide whitespace changes
Inline Side-by-side
Showing with 834 additions and 0 deletions
Report.pdf 0 → 100644
View file @ e4290b58
File added
registration-and-discovery-zookeeper/.idea/.gitignore 0 → 100644
View file @ e4290b58
# Default ignored files
/shelf/
/workspace.xml
registration-and-discovery-zookeeper/.idea/artifacts/Service_Registration_and_Discovery_jar.xml 0 → 100644
View file @ e4290b58
<component name="ArtifactManager">
<artifact type="jar" name="Service-Registration-and-Discovery:jar">
<output-path>$PROJECT_DIR$/out/artifacts/Service_Registration_and_Discovery_jar</output-path>
<root id="archive" name="Service-Registration-and-Discovery.jar">
<element id="module-output" name="Service-Registration-and-Discovery" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-buffer/4.1.94.Final/netty-buffer-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-osx-aarch_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/ch/qos/logback/logback-classic/1.2.10/logback-classic-1.2.10.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/apache/zookeeper/zookeeper-jute/3.9.1/zookeeper-jute-3.9.1.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-linux-x86_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport/4.1.94.Final/netty-transport-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-epoll/4.1.94.Final/netty-transport-native-epoll-4.1.94.Final-linux-x86_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-classes-epoll/4.1.94.Final/netty-transport-classes-epoll-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/ch/qos/logback/logback-core/1.2.10/logback-core-1.2.10.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/commons-io/commons-io/2.11.0/commons-io-2.11.0.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-classes/2.0.61.Final/netty-tcnative-classes-2.0.61.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-handler/4.1.94.Final/netty-handler-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-linux-aarch_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/slf4j/slf4j-api/1.7.30/slf4j-api-1.7.30.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/apache/yetus/audience-annotations/0.12.0/audience-annotations-0.12.0.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-unix-common/4.1.94.Final/netty-transport-native-unix-common-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-windows-x86_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-common/4.1.94.Final/netty-common-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec/4.1.94.Final/netty-codec-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-osx-x86_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver/4.1.94.Final/netty-resolver-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/apache/zookeeper/zookeeper/3.9.1/zookeeper-3.9.1.jar" path-in-jar="/" />
</root>
</artifact>
</component>
\ No newline at end of file
registration-and-discovery-zookeeper/.idea/artifacts/Service_Registration_and_Discovery_jar2.xml 0 → 100644
View file @ e4290b58
<component name="ArtifactManager">
<artifact type="jar" name="Service-Registration-and-Discovery:jar2">
<output-path>$PROJECT_DIR$/out/artifacts/Service_Registration_and_Discovery_jar2</output-path>
<root id="archive" name="Service-Registration-and-Discovery.jar">
<element id="module-output" name="Service-Registration-and-Discovery" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-buffer/4.1.94.Final/netty-buffer-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-osx-aarch_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/ch/qos/logback/logback-classic/1.2.10/logback-classic-1.2.10.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/apache/zookeeper/zookeeper-jute/3.9.1/zookeeper-jute-3.9.1.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-linux-x86_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport/4.1.94.Final/netty-transport-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-epoll/4.1.94.Final/netty-transport-native-epoll-4.1.94.Final-linux-x86_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-classes-epoll/4.1.94.Final/netty-transport-classes-epoll-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/ch/qos/logback/logback-core/1.2.10/logback-core-1.2.10.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/commons-io/commons-io/2.11.0/commons-io-2.11.0.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-classes/2.0.61.Final/netty-tcnative-classes-2.0.61.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-handler/4.1.94.Final/netty-handler-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-linux-aarch_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/slf4j/slf4j-api/1.7.30/slf4j-api-1.7.30.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/apache/yetus/audience-annotations/0.12.0/audience-annotations-0.12.0.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-unix-common/4.1.94.Final/netty-transport-native-unix-common-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-windows-x86_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-common/4.1.94.Final/netty-common-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec/4.1.94.Final/netty-codec-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-tcnative-boringssl-static/2.0.61.Final/netty-tcnative-boringssl-static-2.0.61.Final-osx-x86_64.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver/4.1.94.Final/netty-resolver-4.1.94.Final.jar" path-in-jar="/" />
<element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/apache/zookeeper/zookeeper/3.9.1/zookeeper-3.9.1.jar" path-in-jar="/" />
</root>
</artifact>
</component>
\ No newline at end of file
registration-and-discovery-zookeeper/.idea/compiler.xml 0 → 100644
View file @ e4290b58
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="CompilerConfiguration">
<annotationProcessing>
<profile name="Maven default annotation processors profile" enabled="true">
<sourceOutputDir name="target/generated-sources/annotations" />
<sourceTestOutputDir name="target/generated-test-sources/test-annotations" />
<outputRelativeToContentRoot value="true" />
<module name="Service-Registration-and-Discovery" />
</profile>
</annotationProcessing>
</component>
</project>
\ No newline at end of file
registration-and-discovery-zookeeper/.idea/encodings.xml 0 → 100644
View file @ e4290b58
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Encoding">
<file url="file://$PROJECT_DIR$/src/main/java" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/src/main/resources" charset="UTF-8" />
</component>
</project>
\ No newline at end of file
registration-and-discovery-zookeeper/.idea/jarRepositories.xml 0 → 100644
View file @ e4290b58
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="RemoteRepositoriesConfiguration">
<remote-repository>
<option name="id" value="central" />
<option name="name" value="Central Repository" />
<option name="url" value="https://repo.maven.apache.org/maven2" />
</remote-repository>
<remote-repository>
<option name="id" value="central" />
<option name="name" value="Maven Central repository" />
<option name="url" value="https://repo1.maven.org/maven2" />
</remote-repository>
<remote-repository>
<option name="id" value="jboss.community" />
<option name="name" value="JBoss Community repository" />
<option name="url" value="https://repository.jboss.org/nexus/content/repositories/public/" />
</remote-repository>
</component>
</project>
\ No newline at end of file
registration-and-discovery-zookeeper/.idea/misc.xml 0 → 100644
View file @ e4290b58
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ExternalStorageConfigurationManager" enabled="true" />
<component name="MavenProjectsManager">
<option name="originalFiles">
<list>
<option value="$PROJECT_DIR$/pom.xml" />
</list>
</option>
</component>
<component name="ProjectRootManager" version="2" languageLevel="JDK_17" default="true" project-jdk-name="17 (2)" project-jdk-type="JavaSDK">
<output url="file://$PROJECT_DIR$/../" />
</component>
</project>
\ No newline at end of file
registration-and-discovery-zookeeper/.idea/vcs.xml 0 → 100644
View file @ e4290b58
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
</component>
</project>
\ No newline at end of file
registration-and-discovery-zookeeper/README.md 0 → 100644
View file @ e4290b58
# Cluster Management, Registration and Discovery
*topics*
* [Logback vs SLF4J vs Log4J2 - what is the difference?](https://www.youtube.com/watch?v=SWHYrCXIL38&ab_channel=JavaBrains)
* Introduction to Service Registry & Service Discovery
* Implement Service Registry with Zookeeper
* Integrate into our leader-worker architecture
\ No newline at end of file
registration-and-discovery-zookeeper/pom.xml 0 → 100644
View file @ e4290b58
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>Service-Registration-and-Discovery</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.11.0</version>
<configuration>
<source>17</source>
<target>17</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.9.1</version>
</dependency>
</dependencies>
</project>
registration-and-discovery-zookeeper/src/main/java/Application.java 0 → 100644
View file @ e4290b58
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.IOException;
public class Application implements Watcher {
private static final String address = "192.168.181.136:2181";
private static final int SESSION_TIMEOUT = 3000; //dead client
private static final int DEFAULT_PORT = 8080;
private ZooKeeper zooKeeper;
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
int numberOfWorkers = 0;
if(args.length == 1){
numberOfWorkers = Integer.parseInt(args[0]);
}
else{
System.err.println("Please provide number of workers as an argument.");
System.exit(1);
}
Application application = new Application();
ZooKeeper zooKeeper = application.connectToZookeeper();
ServiceRegistry serviceRegistry = new ServiceRegistry(zooKeeper);
Autohealer autohealer = new Autohealer(numberOfWorkers, "registration-and-discovery-zookeeper/out/artifacts/Service_Registration_and_Discovery_jar2/Service-Registration-and-Discovery.jar",serviceRegistry, zooKeeper);
OnElectionAction onElectionAction = new OnElectionAction(serviceRegistry,autohealer);
LeaderElection leaderElection = new LeaderElection(zooKeeper, onElectionAction);
leaderElection.volunteerForLeadership();
leaderElection.reelectLeader();
application.run();
application.close();
}
public ZooKeeper connectToZookeeper() throws IOException {
this.zooKeeper = new ZooKeeper(address, SESSION_TIMEOUT, this);
return zooKeeper;
}
public void run() throws InterruptedException {
synchronized (zooKeeper) {
zooKeeper.wait();
}
}
private void close() throws InterruptedException {
this.zooKeeper.close();
}
@Override
public void process(WatchedEvent watchedEvent) {
switch (watchedEvent.getType()) {
case None:
if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
System.out.println("Successfully connected to Zookeeper");
} else if (watchedEvent.getState() == Event.KeeperState.Disconnected) {
synchronized (zooKeeper) {
System.out.println("Disconnected from Zookeeper");
zooKeeper.notifyAll();
}
} else if (watchedEvent.getState() == Event.KeeperState.Closed) {
System.out.println("Closed Successfully");
}
break;
}
}
}
registration-and-discovery-zookeeper/src/main/java/Application1.java 0 → 100644
View file @ e4290b58
/*
* MIT License
*
* Copyright (c) 2019 Michael Pogrebinsky - Distributed Systems & Cloud Computing with Java
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
public class Application1 {
public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
if (args.length < 1) {
System.err.println("Please provide the worker address as an argument.");
System.exit(1);
}
String workerAddress = args[0];
Worker worker = new Worker();
worker.connectToZookeeper();
worker.work(workerAddress);
}
}
registration-and-discovery-zookeeper/src/main/java/Autohealer.java 0 → 100644
View file @ e4290b58
/*
* MIT License
*
* Copyright (c) 2019 Michael Pogrebinsky - Distributed Systems & Cloud Computing with Java
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.*;
import static org.apache.commons.io.IOUtils.close;
public class Autohealer implements Watcher {
private static final Logger logger = LoggerFactory.getLogger(Autohealer.class);
private static final String AUTOHEALER_ZNODES_PATH = "/workers";
private ServiceRegistry serviceRegistry;
private final String pathToProgram;
private final int numberOfWorkers;
private ZooKeeper zooKeeper;
private Map<String, Integer> nodesCountMap = new HashMap<>();
public Autohealer(int numberOfWorkers, String pathToProgram, ServiceRegistry serviceRegistry, ZooKeeper zooKeeper) {
this.numberOfWorkers = numberOfWorkers;
this.pathToProgram = pathToProgram;
this.serviceRegistry = serviceRegistry;
this.zooKeeper = zooKeeper;
}
public void startWatchingWorkers() throws KeeperException, InterruptedException {
if (zooKeeper.exists(AUTOHEALER_ZNODES_PATH, this) == null) {
zooKeeper.create(AUTOHEALER_ZNODES_PATH, new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
launchWorkersIfNecessary();
}
public void run() throws InterruptedException {
synchronized (this) {
try {
wait();
} catch (InterruptedException e) {
System.out.println(e.getMessage());
}
}
}
public synchronized void closeZookeeper() throws InterruptedException {
synchronized (this) {
try {
close();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
@Override
public void process(WatchedEvent event) {
switch (event.getType()) {
case None:
if (event.getState() == Event.KeeperState.SyncConnected) {
System.out.println("Successfully connected to Zookeeper");
logger.info("Successfully connected to Zookeeper");
} else {
System.out.println("Disconnected from Zookeeper event");
logger.warn("Disconnected from Zookeeper event");
zooKeeper.notifyAll();
}
break;
case NodeChildrenChanged:
System.out.println("change");
if(event.getPath().equals(AUTOHEALER_ZNODES_PATH)){
try {
launchWorkersIfNecessary();
} catch (KeeperException | InterruptedException e) {
System.out.println("Failed to launch workers: " + e.getMessage());
logger.error("Failed to launch workers: " + e.getMessage());
}
}
break;
}
}
private synchronized void launchWorkersIfNecessary() throws KeeperException, InterruptedException {
List<String> workerZnodes = zooKeeper.getChildren(AUTOHEALER_ZNODES_PATH, this);
int existingWorkers = workerZnodes.size();
nodesCountMap.clear();
for (String child : workerZnodes) {
String znodePath = AUTOHEALER_ZNODES_PATH + "/" + child;
byte[] data = zooKeeper.getData(znodePath, false, null);
String workerAddress = new String(data);
nodesCountMap.put(workerAddress, nodesCountMap.getOrDefault(workerAddress, 0) + 1);
}
List<String> allServiceAddresses = serviceRegistry.getAllServiceAddresses();
for (String address : allServiceAddresses) {
if (!nodesCountMap.containsKey(address)) {
nodesCountMap.put(address, 0);
}
}
System.out.println("existingWorkers: " + existingWorkers);
logger.info("Existing workers: {}", existingWorkers);
int numNodes = allServiceAddresses.size();
int averageWorkersPerNode = 0;
int remainingWorkers = 0;
if(numNodes != 0){
averageWorkersPerNode= numberOfWorkers / numNodes;
remainingWorkers = numberOfWorkers % numNodes;
}
Map<String, Integer> workersToAdd = new HashMap<>();
for (String address : allServiceAddresses) {
int currentWorkers = nodesCountMap.getOrDefault(address, 0);
int workersNeeded = Math.max(0, averageWorkersPerNode - currentWorkers);
workersToAdd.put(address, workersNeeded);
}
List<String> sortedNodes = new ArrayList<>(allServiceAddresses);
sortedNodes.sort(Comparator.comparingInt(node -> nodesCountMap.getOrDefault(node, 0)));
int nodeIndex = 0;
for (int i = 0; i < remainingWorkers; i++) {
String nodeAddress = sortedNodes.get(nodeIndex);
workersToAdd.put(nodeAddress, workersToAdd.get(nodeAddress) + 1);
nodeIndex = (nodeIndex + 1) % numNodes;
}
int workersToLaunch = 0;
if (existingWorkers < numberOfWorkers) {
workersToLaunch = numberOfWorkers - existingWorkers;
}
System.out.println("workersToLaunch: " + workersToLaunch);
logger.info("Workers to launch: {}", workersToLaunch);
System.out.println("Current Nodes Status On Cluster: ");
for (Map.Entry<String, Integer> entry : nodesCountMap.entrySet()) {
System.out.println(entry.getKey() + ": " + entry.getValue());
}
logger.info("Current Nodes Status On Cluster:");
for (Map.Entry<String, Integer> entry : nodesCountMap.entrySet()) {
logger.info("{}: {}", entry.getKey(), entry.getValue());
}
for (String address : sortedNodes) {
int myWorkersToLaunch = workersToAdd.get(address);
for (int i = 0; i < myWorkersToLaunch && workersToLaunch> 0; i++) {
try {
startNewWorker(address, pathToProgram);
workersToLaunch--;
nodesCountMap.put(address, nodesCountMap.getOrDefault(address, 0) + 1);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
System.out.println("Nodes Status After Launching Process: ");
for (Map.Entry<String, Integer> entry : nodesCountMap.entrySet()) {
System.out.println(entry.getKey() + ": " + entry.getValue());
}
logger.info("Nodes Status After Launching Process:");
for (Map.Entry<String, Integer> entry : nodesCountMap.entrySet()) {
logger.info("{}: {}", entry.getKey(), entry.getValue());
}
Thread.sleep(30000);
}
/**
* Helper method to start a single worker
* @throws IOException
*/
private void startNewWorker(String workerAddress, String workerPath) throws IOException, InterruptedException {
String[] addressParts = workerAddress.split("@");
String username = addressParts[0];
String ipAddress = addressParts[1];
String sshCommand = "ssh " + username + "@" + ipAddress + " \"java -jar /home/" + username + "/registration-and-discovery-zookeeper/out/artifacts/Service_Registration_and_Discovery_jar2/Service-Registration-and-Discovery.jar " + username + "@" + ipAddress + "\"";
System.out.println(String.format("Launching worker instance on: %s ", username + "@" + ipAddress));
logger.info("Launching worker instance on: {}", username + "@" + ipAddress);
Runtime.getRuntime().exec(sshCommand);
}
}
registration-and-discovery-zookeeper/src/main/java/LeaderElection.java 0 → 100644
View file @ e4290b58
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.List;
public class LeaderElection implements Watcher {
private static final String ELECTION_NAMESPACE = "/election";
private static final Logger logger = LoggerFactory.getLogger(LeaderElection.class);
private String currentZnodeName;
private ZooKeeper zooKeeper;
private OnElectionCallback onElectionCallback;
public LeaderElection(ZooKeeper zooKeeper, OnElectionCallback onElectionCallback) {
this.zooKeeper = zooKeeper;
this.onElectionCallback = onElectionCallback;
}
public void volunteerForLeadership() throws InterruptedException, KeeperException {
String znodePrefix = ELECTION_NAMESPACE + "/c_";
String znodeFullPath = zooKeeper.create(znodePrefix, new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(znodeFullPath);
logger.info("Znode created: {}", znodeFullPath);
this.currentZnodeName = znodeFullPath.replace(ELECTION_NAMESPACE + "/", "");
}
public void reelectLeader() throws InterruptedException, KeeperException {
String predecessorName = "";
Stat predecessorStat = null;
//this while to guarantee get predecessor even if it deleted just before zookeeper.exist
while (predecessorStat == null) {
List<String> children = zooKeeper.getChildren(ELECTION_NAMESPACE, false);
Collections.sort(children);
String smallestChild = children.get(0); //the first element
if (smallestChild.equals(currentZnodeName)) {
System.out.println("I'm a leader");
logger.info("I'm a leader");
onElectionCallback.onElectedToBeLeader();
return;
} else {
System.out.println("I'm not a leader");
logger.info("I'm not a leader");
int predecessorIndex = children.indexOf(currentZnodeName) - 1;
predecessorName = children.get(predecessorIndex);
predecessorStat = zooKeeper.exists(ELECTION_NAMESPACE + "/" + predecessorName, this);
}
}
onElectionCallback.onWorker();
System.out.println("Watching znode " + predecessorName);
logger.info("Watching znode: {}", predecessorName);
System.out.println();
}
@Override
public void process(WatchedEvent watchedEvent) {
switch (watchedEvent.getType()) {
case NodeDeleted:
try {
reelectLeader();
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (KeeperException e) {
throw new RuntimeException(e);
}
break;
}
}
}
registration-and-discovery-zookeeper/src/main/java/OnElectionAction.java 0 → 100644
View file @ e4290b58
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
public class OnElectionAction implements OnElectionCallback {
private final ServiceRegistry serviceRegistry;
private final Autohealer autohealer;
public OnElectionAction(ServiceRegistry serviceRegistry, Autohealer autohealer) {
this.serviceRegistry = serviceRegistry;
this.autohealer = autohealer;
}
@Override
public void onElectedToBeLeader() {
serviceRegistry.unregisterFromCluster();
serviceRegistry.registerForUpdates(autohealer);
}
@Override
public void onWorker() {
try {
String username = System.getProperty("user.name");
String ipAddress = InetAddress.getLocalHost().getHostAddress();
String currentServerAddress = String.format("%s@%s", username, ipAddress);
serviceRegistry.registerToCluster(currentServerAddress);
serviceRegistry.registerToCluster(currentServerAddress);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (UnknownHostException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
}
registration-and-discovery-zookeeper/src/main/java/OnElectionCallback.java 0 → 100644
View file @ e4290b58
public interface OnElectionCallback {
void onElectedToBeLeader();
void onWorker();
}
registration-and-discovery-zookeeper/src/main/java/ServiceRegistry.java 0 → 100644
View file @ e4290b58
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
public class ServiceRegistry implements Watcher {
private static final String REGISTRY_ZNODE = "/service_registry";
private final Logger logger = LoggerFactory.getLogger(ServiceRegistry.class);
private final ZooKeeper zooKeeper;
private String currentZnode = null;
private List<String> allServiceAddresses = null;
public List<String> getAllServiceAddresses() {
return allServiceAddresses;
}
public ServiceRegistry(ZooKeeper zooKeeper) {
this.zooKeeper = zooKeeper;
createServiceRegistryZnode();
}
private void createServiceRegistryZnode() {
try {
if (zooKeeper.exists(REGISTRY_ZNODE, false) == null) {
zooKeeper.create(REGISTRY_ZNODE, new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
e.printStackTrace();
logger.error("Error creating service registry znode: {}", e.getMessage());
} catch (InterruptedException e) {
e.printStackTrace();
logger.error("Interrupted while creating service registry znode: {}", e.getMessage());
}
}
public void registerToCluster(String metadata) throws KeeperException, InterruptedException {
if (this.currentZnode != null) {
System.out.println("Already registered to service registry");
logger.info("Already registered to service registry");
return;
}
this.currentZnode = zooKeeper.create(REGISTRY_ZNODE + "/n_", metadata.getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("Registered to service registry");
logger.info("Registered to service registry");
}
public void registerForUpdates(Autohealer autohealer) {
try {
updateAddresses(autohealer);
} catch (KeeperException e) {
e.printStackTrace();
logger.error("Error registering for updates: {}", e.getMessage());
} catch (InterruptedException e) {
e.printStackTrace();
logger.error("Interrupted while registering for updates: {}", e.getMessage());
}
}
public void unregisterFromCluster() {
try {
if (currentZnode != null && zooKeeper.exists(currentZnode, false) != null) {
zooKeeper.delete(currentZnode, -1);
}
} catch (KeeperException e) {
e.printStackTrace();
logger.error("Error unregistering from cluster: {}", e.getMessage());
} catch (InterruptedException e) {
e.printStackTrace();
logger.error("Interrupted while unregistering from cluster: {}", e.getMessage());
}
}
Autohealer autohealer1;
private synchronized void updateAddresses(Autohealer autohealer) throws KeeperException, InterruptedException {
autohealer1 = autohealer;
List<String> workerZnodes = zooKeeper.getChildren(REGISTRY_ZNODE, this);
List<String> addresses = new ArrayList<>(workerZnodes.size());
for (String workerZnode : workerZnodes) {
String workerFullPath = REGISTRY_ZNODE + "/" + workerZnode;
Stat stat = zooKeeper.exists(workerFullPath, false);
if (stat == null) {
continue;
}
byte[] addressBytes = zooKeeper.getData(workerFullPath, false, stat);
String address = new String(addressBytes);
addresses.add(address);
}
this.allServiceAddresses = Collections.unmodifiableList(addresses);
System.out.println("The cluster addresses are: " + this.allServiceAddresses);
logger.info("The cluster addresses are: {}", this.allServiceAddresses);
autohealer.startWatchingWorkers();
}
@Override
public void process(WatchedEvent watchedEvent) {
try {
updateAddresses(autohealer1);
} catch (KeeperException e) {
logger.error("Error processing watched event: {}", e.getMessage());
throw new RuntimeException(e);
} catch (InterruptedException e) {
logger.error("Interrupted while processing watched event: {}", e.getMessage());
throw new RuntimeException(e);
}
}
}
registration-and-discovery-zookeeper/src/main/java/Worker.java 0 → 100644
View file @ e4290b58
/*
* MIT License
*
* Copyright (c) 2019 Michael Pogrebinsky - Distributed Systems & Cloud Computing with Java
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.locks.LockSupport;
public class Worker {
private static final String ZOOKEEPER_ADDRESS = "192.168.181.136:2181";
private static final int SESSION_TIMEOUT = 3000;
// Parent Znode where each worker stores an ephemeral child to indicate it is alive
private static final String AUTOHEALER_ZNODES_PATH = "/workers";
private static final float CHANCE_TO_FAIL = 0.01F;
private final Random random = new Random();
private ZooKeeper zooKeeper;
public void connectToZookeeper() throws IOException {
this.zooKeeper = new ZooKeeper(ZOOKEEPER_ADDRESS, SESSION_TIMEOUT, event -> {
});
}
public void work(String workeraddress) throws KeeperException, InterruptedException {
addChildZnode(workeraddress);
while (true) {
System.out.println("Working...");
Thread.sleep(2000);
if (random.nextFloat() < CHANCE_TO_FAIL) {
System.out.println("Critical error happened");
throw new RuntimeException("Oops");
}
}
}
private void addChildZnode(String workeraddress) throws KeeperException, InterruptedException {
String createdZnode= zooKeeper.create(AUTOHEALER_ZNODES_PATH + "/worker_",
workeraddress.getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
}
}
registration-and-discovery-zookeeper/src/main/resources/logback.xml 0 → 100644
View file @ e4290b58
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</Console>
<File name="File" fileName="d:/logfile.log">
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</File>
</Appenders>
<Loggers>
<Root level="WARN">
<AppenderRef ref="File"/>
</Root>
</Loggers>
</Configuration>
\ No newline at end of file
registration-and-discovery-zookeeper/target/classes/logback.xml 0 → 100644
View file @ e4290b58
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</Console>
<File name="File" fileName="d:/logfile.log">
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</File>
</Appenders>
<Loggers>
<Root level="WARN">
<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