From 3426556541256f93d2cba65df3b9c8d1d1772861 Mon Sep 17 00:00:00 2001 From: platania Date: Thu, 16 Feb 2017 11:20:22 -0500 Subject: Initial OpenECOMP Demo commit Change-Id: Ibf8696196a7ac2c84ac8aa7cde1982c9c89fb64d Signed-off-by: platania --- vnfs/vLB/DNSManager/src/main/.DS_Store | Bin 0 -> 6148 bytes vnfs/vLB/DNSManager/src/main/java/.DS_Store | Bin 0 -> 6148 bytes .../src/main/java/DNSMembershipManager.java | 95 ++++++++++++ vnfs/vLB/DNSManager/src/main/java/FDServer.java | 161 +++++++++++++++++++++ 4 files changed, 256 insertions(+) create mode 100644 vnfs/vLB/DNSManager/src/main/.DS_Store create mode 100644 vnfs/vLB/DNSManager/src/main/java/.DS_Store create mode 100644 vnfs/vLB/DNSManager/src/main/java/DNSMembershipManager.java create mode 100644 vnfs/vLB/DNSManager/src/main/java/FDServer.java (limited to 'vnfs/vLB/DNSManager/src/main') 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 Binary files /dev/null and b/vnfs/vLB/DNSManager/src/main/.DS_Store differ 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 Binary files /dev/null and b/vnfs/vLB/DNSManager/src/main/java/.DS_Store differ 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 active = new HashSet(); + + @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 active = new HashSet(); + while(true) { + Set alive_this_round = fd.getAliveProcesses(); + Iterator 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 alive = new ConcurrentHashMap(); // Key: process IP address; Value: # consecutive missed pings + private Set alive_this_round = ConcurrentHashMap.newKeySet(); // Needs to be synchronized because it is accessed by multiple threads + private Set dead = new HashSet(); + + 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 alive_processes = alive.keySet(); + Iterator 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 getAliveProcesses() { + return alive.keySet(); + } +} -- cgit 1.2.3-korg