Add Announce functionality

Author: MoeweX

src/main/java/communication/DirectMessageReceiver.java

@@ -1,12 +1,17 @@
 package communication;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
 import org.apache.log4j.Logger;
 import org.zeromq.ZMQ;
 import org.zeromq.ZMQ.Socket;
 
 import control.FBase;
 import crypto.CryptoProvider.EncryptionAlgorithm;
 import de.hasenburg.fbase.model.GetMissedMessageResponse;
+import exceptions.FBaseEncryptionException;
 import exceptions.FBaseException;
 import model.JSONable;
 import model.config.NodeConfig;
@@ -53,8 +58,10 @@ protected void interpreteReceivedEnvelope(Envelope envelope, Socket responseSock
 
 			envelope.getMessage().decryptFields(fBase.configuration.getPrivateKey(),
 					EncryptionAlgorithm.RSA);
-			envelope.getMessage().verifyMessage(requestingNode.getPublicKey(),
-					EncryptionAlgorithm.RSA);
+			if (!envelope.getMessage().verifyMessage(requestingNode.getPublicKey(),
+					EncryptionAlgorithm.RSA)) {
+				throw new FBaseEncryptionException("The message was not signed correctly");
+			}
 
 			// INTERPRET MESSAGE
 			// if slow, it might be wise to create another task which executes the processing
@@ -74,6 +81,17 @@ protected void interpreteReceivedEnvelope(Envelope envelope, Socket responseSock
 				} catch (FBaseException e) {
 					responseMessage.setTextualInfo("Required messageID not parseable");
 				}
+			} else if (Command.ANNOUNCE_OWN_NODE_CONFIGURATION_CHANGE.equals(envelope.getMessage().getCommand())) { 
+				responseMessage.setContent("true");
+				try {
+					boolean status = fBase.taskmanager.runAnnounceUpdateOfOwnNodeConfigurationTask().get(5, TimeUnit.SECONDS);
+					if (status == false) {
+						responseMessage.setContent("Announcing failed, check log of target machine");
+					}
+				} catch (InterruptedException | ExecutionException | TimeoutException e) {
+					responseMessage.setContent(e.getMessage());
+				}
+				
 			} else {
 				responseMessage.setTextualInfo(
 						"Unknown command " + envelope.getMessage().getCommand().toString());

src/main/java/communication/DirectMessageSender.java

@@ -19,6 +19,7 @@
 import model.messages.Command;
 import model.messages.Envelope;
 import model.messages.Message;
+import tasks.AnnounceUpdateOfOwnNodeConfigurationTaskTest;
 
 /**
  * Sends requests to designated receivers.
@@ -35,13 +36,22 @@ public class DirectMessageSender extends AbstractSender {
 
 	/**
 	 * Initializes the Message, it then can be used without further modifications.
+	 * Messages are send to a random machine of the given node configuration.
 	 */
 	public DirectMessageSender(NodeConfig targetNode, FBase fBase) {
 		super(getRandomAddress(targetNode.getMachines()), targetNode.getMessagePort(), ZMQ.REQ);
 		this.fBase = fBase;
 		this.targetNode = targetNode;
 	}
 
+	/**
+	 * Initializes the Message, it then can be used without further modifications.
+	 */
+	public DirectMessageSender(String targetAddress, int targetPort, FBase fBase) {
+		super(targetAddress, targetPort, ZMQ.REQ);
+		this.fBase = fBase;
+	}
+	
 	private static String getRandomAddress(List<String> machines) {
 		int randomNum = ThreadLocalRandom.current().nextInt(0, machines.size());
 		return "tcp://" + machines.get(randomNum);
@@ -125,6 +135,35 @@ public GetMissedMessageResponse sendGetDataRecord(MessageID messageID)
 		}
 	}
 
+	/**
+	 * Ask the target machine to run {@link AnnounceUpdateOfOwnNodeConfigurationTaskTest}.
+	 * This is necessary if a new machine was added to a node, because node configuration
+	 * updates are distributed via the publisher to other nodes (and other nodes have not
+	 * subscribed to the new machine yet).
+	 * 
+	 * @throws FBaseCommunicationException - if other node declines/cannot be reached
+	 */
+	public void sendAnnounceMeRequest() throws FBaseCommunicationException {
+		Message m = new Message();
+		m.setContent(Command.ANNOUNCE_OWN_NODE_CONFIGURATION_CHANGE.toString());
+		m.setCommand(Command.ANNOUNCE_OWN_NODE_CONFIGURATION_CHANGE);
+		try {
+			String answer = send(createEncryptedEnvelope(m, fBase.configuration.getPublicKey()), null, null);
+			Message response = createDecryptedMessage(answer, fBase.configuration.getPublicKey());
+
+			if (response.getContent().equals("true")) {
+				logger.debug("The other machine announced me");
+				return;
+			} else {
+				logger.debug("The other machine could not announce me");
+				throw new FBaseCommunicationException("The other machine could not announce me " + response.getContent());
+			}
+		} catch (FBaseEncryptionException e1) {
+			logger.error(e1.getMessage(), e1);
+			throw new FBaseCommunicationException("Announce Request failed " + e1.getMessage());
+		}
+	}
+
 	/**
 	 * Create an envelope that is signed with the private key of the node and encrypted with
 	 * the public key of the target node. Also sets

src/main/java/control/FBase.java

@@ -1,19 +1,24 @@
 package control;
 
+import java.util.Iterator;
+import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import org.apache.log4j.Logger;
+import org.javatuples.Pair;
 
 import communication.DirectMessageReceiver;
+import communication.DirectMessageSender;
 import communication.MessageIdEvaluator;
 import communication.NamingServiceSender;
 import communication.Publisher;
 import communication.SubscriptionRegistry;
 import crypto.CryptoProvider.EncryptionAlgorithm;
 import de.hasenburg.fbase.rest.WebServer;
 import exceptions.FBaseCommunicationException;
+import exceptions.FBaseException;
 import exceptions.FBaseNamingServiceException;
 import exceptions.FBaseStorageConnectorException;
 import model.config.ClientConfig;
@@ -88,6 +93,7 @@ public void startup(boolean announce, boolean backgroundTasks) throws Interrupte
 
 		// add machine to node
 		if (announce) {
+			Thread.sleep(400); // make sure own heartbeat is in database
 			announceMachineAdditionToNode();
 		}
 
@@ -106,15 +112,35 @@ public void startup(boolean announce, boolean backgroundTasks) throws Interrupte
 
 	private void announceMachineAdditionToNode() throws FBaseStorageConnectorException,
 			FBaseCommunicationException, FBaseNamingServiceException {
-		if (connector.heartbeats_listAll().size() <= 1) {
+		Map<String, Pair<String, Long>> heartbeats = connector.heartbeats_listAll();
+		if (heartbeats.size() <= 1) {
+			logger.debug("We are the first machine of the node");
 			// update myself (must exist before, created by another node)
-			taskmanager.runAnnounceUpdateOfOwnNodeConfigurationTask();
+			taskmanager.runAnnounceUpdateOfOwnNodeConfigurationTask();	
+			// TODO 2: get all keygroups in which node is either replica/trigger node (important for restart)
+			return;
 		}
 
-		
-		// TODO 1: Tell a machine that is already registered about addition (we need a one-to-one
-		// here)
-
+		logger.debug("In total, the node has " + heartbeats.size() + " including myself.");
+		Iterator<String> iterator = heartbeats.keySet().iterator();
+		while (iterator.hasNext()) {
+			String next = iterator.next();
+			if (!configuration.getMachineName().equals(next)) {
+				DirectMessageSender sender = new DirectMessageSender("tcp://" + heartbeats.get(next).getValue0(), configuration.getMessagePort(), this);
+				try {
+					sender.sendAnnounceMeRequest();
+					sender.shutdown();
+					break;
+				} catch (FBaseException e) {
+					logger.debug("Machine " + next + " could not announce me, trying with another one");
+				}
+				sender.shutdown();
+			}
+			// no machine left for announcing
+			if (iterator.hasNext()) {
+				logger.fatal("No machine could announce me, shutting down");
+			}
+		}
 	}
 
 	public void tearDown() {

src/main/java/storageconnector/S3DBConnector.java

@@ -140,6 +140,7 @@ public void deleteBuckets() throws FBaseStorageConnectorException {
 						break;
 					}
 				}
+				logger.debug("Deleting bucket " + bucketName);
 				s3.deleteBucket(bucketName);
 			} catch (AmazonServiceException e) {
 				throw new FBaseStorageConnectorException(e);

src/main/java/tasks/AnnounceUpdateOfOwnNodeConfigurationTask.java

@@ -54,7 +54,7 @@ public Boolean executeFunctionality() {
 						+ e.getMessage());
 			}
 
-			// publish to all keygroups the machine is responsible for
+			// publish to all keygroups the node is responsible for
 			Set<KeygroupID> responsibilities =
 					fBase.connector.keyGroupSubscriberMachines_listAll().keySet();
 			for (KeygroupID keygroupID : responsibilities) {

src/main/java/tasks/background/PollLatestConfigurationDataForResponsibleKeygroupsTask.java

@@ -35,6 +35,9 @@
  * 
  * TODO T: Write test for this task
  * 
+ * TODO 2: It might make sense to ask the naming service about all keygroups this node is responsbile for,
+ * instead of relying on the data from the responsibility table
+ * 
  * @author jonathanhasenburg
  *
  */

src/test/java/communication/NamingServiceSenderTest.java

@@ -48,7 +48,7 @@ public class NamingServiceSenderTest {
 	private static ExecutorService executor;
 	private static FBase fbase;
 	private static final String ownNodeConfigJSONPath =
-			"src/test/resources/NamingServiceSenderTest_NodeConfig.json";
+			"src/test/resources/NamingServiceSenderTest_NodeConfig1.json";
 
 	private static NamingServiceSender localSender = null;
 	private static NamingServiceSender nsSender = null;

src/test/java/tasks/FBaseFactory.java

@@ -13,7 +13,6 @@ public class FBaseFactory {
 	private static final String BASIC_PROPERTIES_FILE = "FBaseFactory_Basic";
 	private static final String NAMING_SERVICE_PROPERTIES_FILE = "FBaseFactory_NamingService";
 
-
 	public static FBase basic(int instanceNumber, boolean announce, boolean backgroundTasks)
 			throws FBaseStorageConnectorException, InterruptedException, ExecutionException,
 			TimeoutException, FBaseCommunicationException, FBaseNamingServiceException {

src/test/resources/NamingServiceSenderTest_NodeConfig.json renamed to src/test/resources/NamingServiceSenderTest_NodeConfig1.json

@@ -12,5 +12,5 @@
     "messagePort": 6001,
     "restPort": 8081,
     "location": "Everywhere",
-    "description": "Machine specified in JUnit tests"
+    "description": "Machine specified in JUnit tests (Node 1)"
 }

src/test/resources/NamingServiceSenderTest_NodeConfig2.json

@@ -0,0 +1,16 @@
+{
+    "version": 1,
+    "nodeID": {
+        "nodeID": "N2"
+    },
+    "publicKey": "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAqMCldz9e44B0I6V916WUX5tHlw0S4M+zUniAxeRq3rHFOtlKA+ARI8MW85sDz8MAMT51HxhcXOc09L76s0Nh2eWYdvFkaTH66BH2/d6QBIjmPOj8ZWM3t+U6CzqhN9wXqNKLj19Aop+cy4KqhfstiQpAb33803Wto/Zm+Gt3XQEpH/IRs5fgrnp73I4LDenosz5t/Xjo3DMG1+w9fx9ccB+BUikgFPHVDM5bHVT3KZG/9F3Qh4sYNXKWgNuLGpn2v6dVEAFZO/QM4UzJES7O0vHt9Dis4XQloCXqQaZ3FpCd8fQV7xnNO33gdyIJZyjEv8O5ucSuVbddsr+63cQ76QIDAQAB",
+    "encryptionAlgorithm": "RSA",
+    "machines": [
+        "m1"
+    ],
+    "publisherPort": 7002,
+    "messagePort": 6002,
+    "restPort": 8082,
+    "location": "Everywhere",
+    "description": "Configuration fitting for Node 2"
+}