5 files changed, 256 insertions, 0 deletions
diff --git a/vnfs/vLB/DNSManager/src/.DS_Store b/vnfs/vLB/DNSManager/src/.DS_Store
new file mode 100644
index 00000000..201b6cf3
--- /dev/null
+++ b/vnfs/vLB/DNSManager/src/.DS_Store
diff --git a/vnfs/vLB/DNSManager/src/main/.DS_Store b/vnfs/vLB/DNSManager/src/main/.DS_Store
new file mode 100644
index 00000000..d0f5e852
--- /dev/null
+++ b/vnfs/vLB/DNSManager/src/main/.DS_Store
diff --git a/vnfs/vLB/DNSManager/src/main/java/.DS_Store b/vnfs/vLB/DNSManager/src/main/java/.DS_Store
new file mode 100644
index 00000000..097132be
--- /dev/null
+++ b/vnfs/vLB/DNSManager/src/main/java/.DS_Store
diff --git a/vnfs/vLB/DNSManager/src/main/java/DNSMembershipManager.java b/vnfs/vLB/DNSManager/src/main/java/DNSMembershipManager.java
new file mode 100644
index 00000000..a597f3e5
--- /dev/null
+++ b/vnfs/vLB/DNSManager/src/main/java/DNSMembershipManager.java
@@ -0,0 +1,95 @@
+
+/*************************************************************************//**
+ *
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ ****************************************************************************/
+
+package main.java;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+public class DNSMembershipManager {
+	/*
+	 * Uses Failure Detector (FD) to keep track of the DNS servers currently active.
+	 * @param port: the port that the FD service listens for incoming UDP packets
+	 * @param timeout: how often the FD checks the status of client processes
+	 * @param threshold: number of missing ping messages before declaring a client dead
+	 * @param debug: debug mode on/off
+	 */
+	
+	static Set<String> active = new HashSet<String>();
+	
+	@SuppressWarnings("static-access")
+	public static void main(String[] args) throws IOException, InterruptedException {
+		if(args.length != 5) {
+			System.out.println("Missing input parameters");
+			System.out.println("Usage:");
+			System.out.print("\t- java FDServer [public IP address] [port] [timeout (sec)] [threshold] [debug]\n");
+			System.exit(0);
+		}
+		
+		// Input parameters: PORT, TIMEOUT, THRESHOLD
+		String IPADDR = args[0];
+		int PORT = Integer.parseInt(args[1]);
+		long TIMEOUT = Long.parseLong(args[2]) * 1000; // convert the FD timeout to milliseconds
+		int THRESHOLD = Integer.parseInt(args[3]);
+		int debug = Integer.parseInt(args[4]);
+		boolean DEBUG;
+		if(debug <= 0)
+			DEBUG = false;
+		else
+			DEBUG = true;
+		
+		// Start Failure Detector
+		FDServer fd = new FDServer(IPADDR, PORT, TIMEOUT, THRESHOLD, DEBUG);
+		
+		// Check the status of client processes periodically. We use the same timeout value as FD
+		Set<String> active = new HashSet<String>();
+		while(true) {
+			Set<String> alive_this_round = fd.getAliveProcesses();
+			Iterator<String> iter = alive_this_round.iterator();
+			String pid;
+			
+			// Check if there is some new DNS active
+			while(iter.hasNext()) {
+				pid = iter.next();
+				if(!active.contains(pid)) {
+					active.add(pid);
+					// Add the new vDNS to the set of vDNS servers
+					String script = new String("bash /opt/FDserver/add_dns.sh " + pid);
+		            Runtime.getRuntime().exec(script);
+					if(DEBUG) System.out.println("Adding process " + pid + " to the list of DNSs");
+				}
+			}
+			
+			// Remove possible dead DNSs
+			iter = active.iterator();
+			while(iter.hasNext()) {
+				pid = iter.next();
+				if(!alive_this_round.contains(pid)) {
+					iter.remove(); // remove element from the iterator to avoid ConcurrentModificationException
+					// Remove the new vDNS from the set of vDNS servers
+					String script = new String("bash /opt/FDserver/remove_dns.sh " + pid);
+		            Runtime.getRuntime().exec(script);
+					if(DEBUG) System.out.println("Removing process " + pid + " from the list of DNSs");
+				}
+			}
+			Thread.currentThread().sleep(TIMEOUT);
+		}
+	}
+}
diff --git a/vnfs/vLB/DNSManager/src/main/java/FDServer.java b/vnfs/vLB/DNSManager/src/main/java/FDServer.java
new file mode 100644
index 00000000..4b64c584
--- /dev/null
+++ b/vnfs/vLB/DNSManager/src/main/java/FDServer.java
@@ -0,0 +1,161 @@
+
+/*************************************************************************//**
+ *
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ ****************************************************************************/
+
+package main.java;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+/*
+ * Server-side Failure Detector (FD) implementation.
+ * @param port: the port that the FD service listens for incoming UDP packets
+ * @param timeout: how often the FD checks the status of client processes
+ * @param threshold: number of missing ping messages before declaring a client dead
+ * @param debug: debug mode on/off
+ */
+
+public class FDServer implements Runnable {
+	// Input parameters
+	private String IP_ADDR;
+	private int PORT;
+	private long TIMEOUT;
+	private int THRESHOLD;
+	private boolean DEBUG;
+	
+	// Data structures that store information about alive processes, processes alive/dead this round
+	private Map<String, Integer> alive = new ConcurrentHashMap<String, Integer>(); // Key: process IP address; Value: # consecutive missed pings
+	private Set<String> alive_this_round = ConcurrentHashMap.newKeySet(); // Needs to be synchronized because it is accessed by multiple threads
+	private Set<String> dead = new HashSet<String>();
+	
+	public FDServer(String ip_addr, int port, long timeout, int threshold, boolean debug) throws IOException {
+		IP_ADDR = ip_addr;
+		PORT = port;
+		TIMEOUT = timeout;
+		THRESHOLD = threshold;
+		DEBUG = debug;
+		(new Thread(this)).start();
+	}
+	
+	@Override
+	public void run() {
+		try {
+			runFD();
+		} catch (IOException e) {
+			e.printStackTrace();
+		}	
+	}
+	
+	private void runFD() throws IOException {
+		// Check the status of client processes periodically
+		TimerTask timer = new TimerTask() {
+			public void run() {
+				checkClientStatus();
+			}
+		};
+		new Timer().scheduleAtFixedRate(timer, 0, TIMEOUT);
+		
+		// Define a DatagramSocket object for receiving incoming UDP packets
+		@SuppressWarnings("resource")
+		DatagramSocket sock = new DatagramSocket(PORT);
+		byte[] buffer = new byte[256];
+		
+		// Wait for incoming PING messages from remote clients
+		while(true) {
+			DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
+            sock.receive(packet);
+            String[] content = new String(packet.getData()).trim().split(":"); // Remove leading and trailing spaces
+            String msg_type = content[0];
+            String pid = content[1];
+            // Process only PING UDP packets
+            if(msg_type.equals("PING")) {
+            	String ip = packet.getAddress().getHostAddress();
+            	alive_this_round.add(pid);
+				if(DEBUG) System.out.println("Keep-alive message received from process " + pid + " (sender IP Address: " + ip +")");
+				sendReplyMessage(packet.getAddress());
+            }
+            else {
+            	if(DEBUG) System.out.println("The received message is not a PING. Received content: " + content);
+            }
+		}
+	}
+	
+	private void sendReplyMessage(InetAddress address) throws IOException {
+		DatagramSocket sock = new DatagramSocket();
+		// Allocate buffer for the PING message
+		String content = "PONG:" + IP_ADDR;
+		byte[] buffer = content.getBytes();
+		// Sent a PONG message
+		DatagramPacket packet = new DatagramPacket(buffer, buffer.length, address, PORT);
+		sock.send(packet);
+		sock.close();
+	}
+
+	// Update the list of processes that are alive
+	private void checkClientStatus() {
+		if(DEBUG) System.out.println("/================================/");
+		if(DEBUG) System.out.println("Update status of remote processes");
+		// Check if a process alive the previous round is still alive
+		// Otherwise increment its counter
+		Set<String> alive_processes = alive.keySet();
+		Iterator<String> iter = alive_processes.iterator();
+		while(iter.hasNext()) {
+			String process = iter.next();
+			if(!alive_this_round.contains(process)) {
+				int counter = alive.get(process) + 1;
+				alive.put(process, counter);
+				if(DEBUG) System.out.println("Process " + process + " hasn't sent a message " + counter + " time(s) in a row");
+				// If the number of consecutive missed ping messages reached the threshold, 
+				// then assume the process to be dead
+				if(counter == THRESHOLD) {
+					dead.add(process);
+					if(DEBUG) System.out.println("Process " + process + " is dead");
+				}
+			}
+		}
+		
+		// Processes alive this round
+		iter = alive_this_round.iterator();
+		while(iter.hasNext()) {
+			String process = iter.next();
+			alive.put(process, 0);
+			if(DEBUG) System.out.println("Process " + process + " is alive this round");
+		}
+			
+		
+		// Remove dead processes
+		iter = dead.iterator();
+		while(iter.hasNext()) {
+			String process = iter.next();
+			if(alive.containsKey(process))
+				alive.remove(process);
+			if(DEBUG) System.out.println("Process " + process + " is removed from the list of alive processes");
+		}
+		
+		// Cleanup
+		alive_this_round.clear();
+		dead.clear();
+		if(DEBUG) System.out.println();
+	}
+	
+	// Return the set of alive processes to up-stream applications
+	public Set<String> getAliveProcesses() {
+		return alive.keySet();
+	}
+}