Merge pull request #22 from OpenFogStack/subscriptionManagement

Add Rest-Client and update subscription management to thesis version

Some tests are missing and S3 Connector needs to be adjusted

Author: MoeweX

ImplementationSubscriptionManagement.md

@@ -0,0 +1,86 @@
+# Implementation Subscription Management
+
+## Events
+
+### [1] Missing heartbeats detected
+
+- [x] Background task that stores own heartbeats in the node database
+- [x] Background task to detect missing heartbeats
+- [x] Detected: run RemoveMachineFromNodeTask
+
+### [2] No responsibility detected
+- [x] Background task to detect missing responsibilities
+- [x] Detected: run UpdateKeygroupSubscriptionsTask
+
+### [3] Lost responsibility detected
+
+- [x] Background task to detect lost responsibilities for each keygroup for which subscriptions exist
+- [x] Detected: run UpdateKeygroupSubscriptionsTask
+
+### [4] Client updates/deletes keygroup via node
+
+- [x] Add client keygroup update/delete functionality incl. naming service validation
+- [x] Naming service approves: run UpdateKeygroupConfigTask
+
+### [5] Not interpretable message
+
+- [x] Add encryption exception handling to subscriber
+- [x] Catched: get newest keygroup config version from naming service
+- [x] Run UpdateKeygroupConfigTask
+
+### [6] Periodic configuration update
+
+- [x] Background task that periodically polls all responsible keygroup configurations and node configurations for nodes present in the keygroups
+- [x] Keygroup: run UpdateKeygroupConfigTask
+- [x] Node: UpdateForeignNodeConfigTask
+
+### [7] Recognize foreign keygroup update
+Needed if [4] is performed by other node
+
+- [x] Add background task that checks whether version of keygroup config changed
+- [x] Changed: Run UpdateKeygroupSubscriptionsTask
+
+### [8] Subscriber receives configuration update
+
+- [x] Add configuration update handling to subscriber
+- [x] Run UpdateKeygroupConfigTask or UpdateForeignNodeConfigTask
+
+## Tasks (not background)
+
+### [D] UpdateKeygroupSubscriptionsTask
+
+1. Remove all subscriptions for a given keygroup
+2. Create new subscriptions if all `true`:
+  - node still apart of the keygroup
+  - keygroup not tombstoned
+  - no machine responsible yet/I am responsible
+3. Update responsibility table
+
+### [B] UpdateKeygroupConfigTask
+Only run if version differs
+
+1. Put configuration in node database
+2. Run UpdateKeygroupSubscriptionsTask
+3. Publish config to all subscribers if started by [4]
+
+### [C] UpdateForeignNodeConfigTask
+Only run if the machines changed
+
+1. Put configuration in node database
+2. Get all keygroups in which node participates
+  * Run UpdateKeygroupSubscriptionsTask for keygroup
+
+### [A] RemoveMachineFromNodeTask
+
+1. Remove machine from responsibility table
+2. Remove machine from heartbeats table
+3. Rebuild nodeconfig and send to naming service
+4. Rebuild nodeconfig and publish to all subscribers for all keygroups
+5. UpdateKeygroupSubscriptionsTask
+
+### AddMachineToNodeTask
+Not used here, is used by startup functionality
+
+1. Put heartbeat into heartbeats
+2. Rebuild nodeconfig and send to naming service
+3. Rebuild nodeconfig and publish to all subscribers for all keygroups (must be done by another node!)

README.md

@@ -9,44 +9,23 @@ For some of the tests, a running naming service is required. The naming service
 # Current ToDos / Missing Functionality
 
 ## Startup
-- [ ] A machineName should be dynamically created on Startup
-- [ ] Background task that checks whether node config contains myself (might have been overwritten by removal of machine due to heartbeats)
-- [ ] Add background task that checks whether node config only contains nodes of machines that have heartbeats (overwritten by starting node)
+- [ ] A machineName should be dynamically created on Startup (clean up configs)
+- [x] Run AddMachineToNodeTask (ended up not being a task)
 
-## General Node Database
-- [x] Add a connector that supports multi-machine nodes
-- [x] Add versions to all configurations so that a node/machine can identify updates
-
-## Publish/Subscribe
-- [x] Add capabilities to process messages that cannot be encrypted
-
-### Subscription Management
-- [x] Add background task that checks whether any of the keygroups I am responsible for have been updated by another machine (CheckKeygroupConfigurationsOnUpdatesTask) #11
-- [x] Instead of unsubscribing/subscribing, each keygroup config update should lead to a complete reset of subscriptions
-- [ ] Make subscription management code as defined in thesis (event detection)
-
-### Heartbeats
-- [ ] Add background task that stores own heartbeats in the node database #11
-- [ ] Add background task that checks other machine's heartbeats and removes machines from a node if they did not respond to long #11 (run subscription management and clean heartbeat table/node config after started subscriptions)
+## Subscription Management incl. Heartbeats
+- See ImplementationSubscriptionManagement.md
 
 ## One to One Communication
-- [x] Rebuild asymmetric encryption so that it uses a symmetric approach for the actual data
 - [ ] Add one to one communication for datarecords
 
-### Naming Service based Management
-- [x] Add keygroup configuration control methods to sender
-- [x] Add response processing to all methods
-- [x] Add node management methods (e.g. to update a node configuration when a machine is added/removed)
-- [x] Add a background task that periodically polls the naming service about the newest configurations
-
 ### Handling Missed Messages
-- [ ] Add message history size to machine config (node specific)
-- [x] Messaging data needs to be stored after each data related publish
-- [x] Add sender/receiver capabilities for missed messages
-- [x] Write and test logic that uses functionality
+- [ ] Add message history size to machine config (node specific), cleanup config files
 - [ ] (Add message history cleanup functionality (on receiver and sender side)), thesis states not implemented
 
 ## Controlling FBase with Clients
- - [ ] Enable asymmetric encryption
- - [ ] Identify and add missing control methods
- - [ ] Keygroup updates/deletions from client are forwarded to other nodes
+ - [ ] Enable encryption and authentication
+ - [ ] Method to instruct machine to update all configurations with naming service data
+ - [x] Keygroup C, R, D
+ - [x] Keygroup all update methods
+ - [X] Client C, D
+ - [X] DataRecord P, R, D, List

pom.xml

@@ -60,14 +60,34 @@
 		</dependency>
 		<dependency>
 			<groupId>org.eclipse.jetty</groupId>
-			<artifactId>jetty-security</artifactId>
+			<artifactId>jetty-servlet</artifactId>
 			<version>9.4.6.v20170531</version>
 		</dependency>
 		<dependency>
 			<groupId>org.eclipse.jetty</groupId>
-			<artifactId>jetty-servlet</artifactId>
+			<artifactId>jetty-util</artifactId>
 			<version>9.4.6.v20170531</version>
 		</dependency>
+		<dependency>
+			<groupId>org.glassfish.jersey.core</groupId>
+			<artifactId>jersey-server</artifactId>
+			<version>2.26</version>
+		</dependency>
+		<dependency>
+			<groupId>org.glassfish.jersey.containers</groupId>
+			<artifactId>jersey-container-servlet-core</artifactId>
+			<version>2.26</version>
+		</dependency>
+		<dependency>
+			<groupId>org.glassfish.jersey.containers</groupId>
+			<artifactId>jersey-container-jetty-http</artifactId>
+			<version>2.26</version>
+		</dependency>
+		<dependency>
+			<groupId>org.glassfish.jersey.inject</groupId>
+			<artifactId>jersey-hk2</artifactId>
+			<version>2.26</version>
+		</dependency>
 		<!-- Client -->
 		<dependency>
 			<groupId>com.mashape.unirest</groupId>

src/main/java/client/Client.java

@@ -1,20 +1,13 @@
 package client;
 
-import java.io.IOException;
-import java.util.List;
-
 import org.apache.log4j.Logger;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.mashape.unirest.http.HttpResponse;
-import com.mashape.unirest.http.Unirest;
 import com.mashape.unirest.http.exceptions.UnirestException;
 
-import model.JSONable;
+import exceptions.FBaseEncryptionException;
 import model.data.DataIdentifier;
 import model.data.DataRecord;
 import model.data.KeygroupID;
-import model.messages.Message;
 
 /**
  * A client implemantation which allows the usage of the FBase rest interface
@@ -25,111 +18,59 @@
 public class Client {
 
 	private static Logger logger = Logger.getLogger(Client.class.getName());
-
-	public static void main(String[] args) throws UnirestException {
-		Client c = new Client();
-		DataRecord record = new DataRecord();
-		DataIdentifier dataIdentifier = new DataIdentifier("smartlight", "h1", "brightness", "M-1");
-		record.setDataIdentifier(new DataIdentifier(dataIdentifier.getKeygroupID(), "M-4"));
-		Message m = c.runPutRecordRequest("http://localhost", 8080, record);
-		logger.info(JSONable.toJSON(m));
-
-		DataRecord record2 = c.runGetRecordRequest("http://localhost", 8080, dataIdentifier);
-		logger.info(JSONable.toJSON(record2));
-
-		List<String> list = c.runGetListRecordRequest("http://localhost", 8080,
-				dataIdentifier.getKeygroupID());
-		list.stream().forEach(i -> logger.info(i));
+	
+	public void getScenario(String address, int port) throws UnirestException {
+		KeygroupID keygroupID = new KeygroupID("smartlight", "h1", "brightness");
+		DataIdentifier dataID = new DataIdentifier(keygroupID, "M-1");
+		
+		DataRecord newRecord = new DataRecord(new DataIdentifier(keygroupID, "M-2"), null);
+		newRecord.setValueWithoutKey("Example value");
+		
+		RecordRequest records = new RecordRequest(address, port);
+		
+		logger.info("Got M-1: " + records.getDataRecord(dataID));
+		logger.info("Put M-2: " + records.putDataRecord(newRecord));
+		logger.info("Got M-2: " + records.getDataRecord(newRecord.getDataIdentifier()));
+		logger.info("List: " + records.listDataRecords(keygroupID));
+		logger.info("Deleted M-2: " + records.deleteDataRecord(newRecord.getDataIdentifier()));
+		logger.info("Got M-2: " + records.getDataRecord(newRecord.getDataIdentifier()));
+		logger.info("List: " + records.listDataRecords(keygroupID));
 	}
-
-	public Client() {
+	
+	public static void main(String[] args) throws UnirestException, FBaseEncryptionException {
+		String address = "localhost";
+		int port = 8081;
+		
+		Client c = new Client();
+		c.getScenario(address, port);
 	}
 
-	public DataRecord runGetRecordRequest(String address, int port, DataIdentifier dataIdentifier)
-			throws UnirestException {
-		DataRecord record = null;
-		try {
-			String target = address + ":" + port + "/record";
-			logger.info("Running get request targeting " + target);
-			HttpResponse<String> response = Unirest.get(target)
-					.queryString("dataIdentifier", dataIdentifier).asString();
-			if (response.getStatus() == 200) {
-				logger.info("Status = 200");
-				Message m = JSONable.fromJSON(response.getBody(), Message.class);
-				// Insert decryption here if needed
-				record = JSONable.fromJSON(m.getContent(), DataRecord.class);
-			} else {
-				logger.info("Status = " + response.getStatus());
-			}
-		} catch (UnirestException e) {
-			e.printStackTrace();
-		}
 
-		return record;
-	}
 
-	@SuppressWarnings("unchecked")
-	public List<String> runGetListRecordRequest(String address, int port, KeygroupID keygroupID) {
-		List<String> list = null;
-		try {
-			String target = address + ":" + port + "/record/list";
-			logger.info("Running get request targeting " + target);
-			HttpResponse<String> response = Unirest.get(target)
-					.queryString("keygroupID", keygroupID).asString();
-			if (response.getStatus() == 200) {
-				logger.info("Status = 200");
-				Message m = JSONable.fromJSON(response.getBody(), Message.class);
-				// Insert decryption here if needed
-				ObjectMapper mapper = new ObjectMapper();
-				list = mapper.readValue(m.getContent(), List.class);
-			} else {
-				logger.info("Status = " + response.getStatus());
-			}
-		} catch (UnirestException | IOException e) {
-			e.printStackTrace();
-		}
-		return list;
-	}
+//	public boolean keygroupConfig_create(String address, int port, KeygroupConfig keygroupConfig)
+//			throws UnirestException, FBaseEncryptionException {
+//		
+//		// prepare
+//		String target = address + ":" + port + "/keygroupConfig";
+//		logger.info("Running post request targeting " + target);
+//		
+//		// create and sign request message
+//		Message requestM = new Message();
+//		requestM.setContent(JSONable.toJSON(keygroupConfig));
+//		requestM.signMessage(privateKey, algorithm);
+//
+//		// send message
+//		HttpResponse<String> response = Unirest.post(target).header("accept", "application/json")
+//				.queryString("clientID", clientID.getID()).body(JSONable.toJSON(requestM))
+//				.asString();
+//
+//		// process response
+//		if (response.getStatus() == 200) {
+//			return true;
+//		} else {
+//			logger.error("Status = " + response.getStatus());
+//			return false;
+//		}
+//	}
 
-	public Message runPutRecordRequest(String address, int port, DataRecord record) {
-		Message m = null;
-		try {
-			String target = address + ":" + port + "/record";
-			logger.info("Running put request targeting " + target);
-			HttpResponse<String> response = Unirest.put(target).header("accept", "application/json")
-					// insert decryption here if needed
-					.queryString("keygroupID", record.getKeygroupID().getID()).body(JSONable.toJSON(record))
-					.asString();
-			if (response.getStatus() == 200) {
-				m = JSONable.fromJSON(response.getBody(), Message.class);
-			} else {
-				logger.info("Status = " + response.getStatus());
-			}
-		} catch (UnirestException e) {
-			e.printStackTrace();
-		}
-		return m;
-	}
-	
-	public Message runDeleteRecordRequest(String address, int port, DataIdentifier identifier) {
-		Message m = null;
-		try {
-			String target = address + ":" + port + "/record";
-			logger.info("Running delete request targeting " + target);
-			HttpResponse<String> response = Unirest.delete(target)
-					  .header("accept", "application/json")
-					  .queryString("keygroupID", identifier.getKeygroupID().getID())
-					  .body(JSONable.toJSON(identifier)) // insert encryption here if needed
-					  .asString();
-			if (response.getStatus() == 200) {
-				m = JSONable.fromJSON(response.getBody(), Message.class);
-			} else {
-				logger.info("Status = " + response.getStatus());
-			}
-		} catch (UnirestException e) {
-			e.printStackTrace();
-		}
-		return m;
-	}
-	
 }