@Test public void testAddNode() throws Exception { waitForFullMesh(2000); IStoreClient<String, String> client0 = syncManagers[0].getStoreClient("global", String.class, String.class); IStoreClient<String, String> client1 = syncManagers[1].getStoreClient("global", String.class, String.class); client0.put("key", "value"); waitForValue(client1, "key", "value", 2000, "client1"); nodes.add(new Node("localhost", 40105, (short)5, (short)5)); SyncManager[] sms = Arrays.copyOf(syncManagers, syncManagers.length + 1); FloodlightModuleContext[] fmcs = Arrays.copyOf(moduleContexts, moduleContexts.length + 1); sms[syncManagers.length] = new SyncManager(); fmcs[moduleContexts.length] = new FloodlightModuleContext(); nodeString = mapper.writeValueAsString(nodes); setupSyncManager(fmcs[moduleContexts.length], sms[syncManagers.length], nodes.get(syncManagers.length)); syncManagers = sms; moduleContexts = fmcs; for(int i = 0; i < 4; i++) { moduleContexts[i].addConfigParam(syncManagers[i], "nodes", nodeString); syncManagers[i].doUpdateConfiguration(); } waitForFullMesh(2000); IStoreClient<String, String> client4 = syncManagers[4].getStoreClient("global", String.class, String.class); client4.put("newkey", "newvalue"); waitForValue(client4, "key", "value", 2000, "client4"); waitForValue(client0, "newkey", "newvalue", 2000, "client0"); }
@LogMessageDocs({ @LogMessageDoc(level="INFO", message="[{id}] Updating sync configuration {config}", explanation="The sync service cluster configuration has been updated"), @LogMessageDoc(level="INFO", message="[{id}] Local node configuration changed; restarting sync" + "service", explanation="The sync service must be restarted to update its configuration") }) protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.debug("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.debug("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }
SyncManager extends AbstractSyncManager { @LogMessageDocs({ @LogMessageDoc(level="INFO", message="[{id}] Updating sync configuration {config}", explanation="The sync service cluster configuration has been updated"), @LogMessageDoc(level="INFO", message="[{id}] Local node configuration changed; restarting sync" + "service", explanation="The sync service must be restarted to update its configuration") }) protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.debug("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.debug("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } } }
SyncManager extends AbstractSyncManager { @LogMessageDocs({ @LogMessageDoc(level="INFO", message="[{id}] Updating sync configuration {config}", explanation="The sync service cluster configuration has been updated"), @LogMessageDoc(level="INFO", message="[{id}] Local node configuration changed; restarting sync" + "service", explanation="The sync service must be restarted to update its configuration") }) protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.debug("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.debug("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }  }
SyncManager extends AbstractSyncManager { @LogMessageDocs({ @LogMessageDoc(level="INFO", message="[{id}] Updating sync configuration {config}", explanation="The sync service cluster configuration has been updated"), @LogMessageDoc(level="INFO", message="[{id}] Local node configuration changed; restarting sync" + "service", explanation="The sync service must be restarted to update its configuration") }) protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.debug("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.debug("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }  @Override void registerStore(String storeName, Scope scope); @Override void registerPersistentStore(String storeName, Scope scope); ClusterConfig getClusterConfig(); void cleanup(); @LogMessageDoc(level="INFO", message="[{id}->{id}] Synchronizing local state to remote node", explanation="Normal state resynchronization is occurring") void antientropy(Node node); void antientropy(); void writeSyncValue(String storeName, Scope scope,
                               boolean persist,
                               byte[] key, Iterable<Versioned<byte[]>> values); boolean handleSyncOffer(String storeName,
                                   byte[] key,
                                   Iterable<VectorClock> versions); IStorageEngine<ByteArray, byte[]> getRawStore(String storeName); IThreadPoolService getThreadPool(); @LogMessageDoc(level="WARN", message="Sync task queue full and not emptying", explanation="The synchronization service is overloaded", recommendation=LogMessageDoc.CHECK_CONTROLLER) void queueSyncTask(SynchronizingStorageEngine e,
                              ByteArray key, Versioned<byte[]> value); @Override void addListener(String storeName, MappingStoreListener listener); void updateConfiguration(); Cursor getCursor(int cursorId); Cursor newCursor(String storeName); void closeCursor(Cursor cursor); @Override IStore<ByteArray,byte[]> getStore(String storeName); @Override short getLocalNodeId(); @Override void shutdown(); @Override void init(FloodlightModuleContext context); @Override void startUp(FloodlightModuleContext context); @Override Collection<Class<? extends IFloodlightService>> getModuleDependencies(); }
SyncManager extends AbstractSyncManager { @LogMessageDocs({ @LogMessageDoc(level="INFO", message="[{id}] Updating sync configuration {config}", explanation="The sync service cluster configuration has been updated"), @LogMessageDoc(level="INFO", message="[{id}] Local node configuration changed; restarting sync" + "service", explanation="The sync service must be restarted to update its configuration") }) protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.debug("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.debug("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }  @Override void registerStore(String storeName, Scope scope); @Override void registerPersistentStore(String storeName, Scope scope); ClusterConfig getClusterConfig(); void cleanup(); @LogMessageDoc(level="INFO", message="[{id}->{id}] Synchronizing local state to remote node", explanation="Normal state resynchronization is occurring") void antientropy(Node node); void antientropy(); void writeSyncValue(String storeName, Scope scope,
                               boolean persist,
                               byte[] key, Iterable<Versioned<byte[]>> values); boolean handleSyncOffer(String storeName,
                                   byte[] key,
                                   Iterable<VectorClock> versions); IStorageEngine<ByteArray, byte[]> getRawStore(String storeName); IThreadPoolService getThreadPool(); @LogMessageDoc(level="WARN", message="Sync task queue full and not emptying", explanation="The synchronization service is overloaded", recommendation=LogMessageDoc.CHECK_CONTROLLER) void queueSyncTask(SynchronizingStorageEngine e,
                              ByteArray key, Versioned<byte[]> value); @Override void addListener(String storeName, MappingStoreListener listener); void updateConfiguration(); Cursor getCursor(int cursorId); Cursor newCursor(String storeName); void closeCursor(Cursor cursor); @Override IStore<ByteArray,byte[]> getStore(String storeName); @Override short getLocalNodeId(); @Override void shutdown(); @Override void init(FloodlightModuleContext context); @Override void startUp(FloodlightModuleContext context); @Override Collection<Class<? extends IFloodlightService>> getModuleDependencies(); static IDebugCounter counterHints; static IDebugCounter counterSentValues; static IDebugCounter counterReceivedValues; static IDebugCounter counterPuts; static IDebugCounter counterGets; static IDebugCounter counterIterators; static IDebugCounter counterErrorRemote; static IDebugCounter counterErrorProcessing; }