Changed the controller registration to use Curator's service discovery package. This removes some bugs that were in the previous implementation. Additionally it's more flexible and extensible, so should we want to register more information that just controller id, this will be easy to do.
diff --git a/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java b/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
index acb6e52..476f7b3 100644
--- a/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
+++ b/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
@@ -1,7 +1,6 @@
package net.onrc.onos.registry.controller;
import java.io.IOException;
-import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -16,9 +15,6 @@
import net.floodlightcontroller.core.module.IFloodlightService;
import net.floodlightcontroller.restserver.IRestApiService;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
import org.openflow.util.HexString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,7 +33,10 @@
import com.netflix.curator.framework.recipes.leader.LeaderLatchListener;
import com.netflix.curator.framework.recipes.leader.Participant;
import com.netflix.curator.retry.ExponentialBackoffRetry;
-import com.netflix.curator.utils.ZKPaths;
+import com.netflix.curator.x.discovery.ServiceCache;
+import com.netflix.curator.x.discovery.ServiceDiscovery;
+import com.netflix.curator.x.discovery.ServiceDiscoveryBuilder;
+import com.netflix.curator.x.discovery.ServiceInstance;
/**
* A registry service that uses Zookeeper. All data is stored in Zookeeper,
@@ -57,11 +56,12 @@
private final String namespace = "onos";
private final String switchLatchesPath = "/switches";
- private final String controllerPath = "/controllers";
+
+ private final String SERVICES_PATH = "/"; //i.e. the root of our namespace
+ private final String CONTROLLER_SERVICE_NAME = "controllers";
protected CuratorFramework client;
- protected PathChildrenCache controllerCache;
protected PathChildrenCache switchCache;
protected ConcurrentHashMap<String, SwitchLeadershipData> switches;
@@ -151,6 +151,8 @@
}
};
+ protected ServiceDiscovery<ControllerService> serviceDiscovery;
+ protected ServiceCache<ControllerService> serviceCache;
@Override
@@ -169,8 +171,6 @@
throw new RegistryException("Already contesting control for " + dpidStr);
}
- log.debug("No latch for switch {}", dpidStr);
-
LeaderLatch latch = new LeaderLatch(client, latchPath, controllerId);
latch.addListener(new SwitchLeaderListener(dpidStr, latch));
@@ -213,8 +213,6 @@
log.debug("Trying to release control of a switch we are not contesting");
return;
}
-
- log.debug("swData for {} was not null: {}", dpidStr, swData);
LeaderLatch latch = swData.getLatch();
@@ -252,17 +250,13 @@
log.debug("Getting all controllers");
List<String> controllers = new ArrayList<String>();
- for (ChildData data : controllerCache.getCurrentData()){
-
- String d = null;
- try {
- d = new String(data.getData(), "UTF-8");
- } catch (UnsupportedEncodingException e) {
- throw new RegistryException("Error encoding string", e);
+ for (ServiceInstance<ControllerService> instance : serviceCache.getInstances()){
+ String id = instance.getPayload().getControllerId();
+ if (!controllers.contains(id)){
+ controllers.add(id);
}
-
- controllers.add(d);
}
+
return controllers;
}
@@ -275,52 +269,20 @@
controllerId = id;
- byte bytes[] = id.getBytes(Charsets.UTF_8);
- String path = ZKPaths.makePath(controllerPath, controllerId);
-
- log.info("Registering controller with id {}", id);
-
try {
- //We need to set a watch to recreate the node in the controller
- //registry if it gets deleted - e.g. on Zookeeper connection loss.
- Watcher watcher = new Watcher(){
- @Override
- public void process(WatchedEvent event) {
- log.debug("got any watch event {} ", event);
-
- String path = ZKPaths.makePath(controllerPath, controllerId);
- byte bytes[] = controllerId.getBytes(Charsets.UTF_8);
-
- try {
- if (event.getType() == Event.EventType.NodeDeleted){
- log.debug("got a node deleted event");
-
-
- client.create().withMode(CreateMode.EPHEMERAL)
- .forPath(path, bytes);
- }
- } catch (Exception e) {
- log.warn("Error recreating controller node for {}: {}",
- controllerId, e.getMessage());
- } finally {
- try {
- client.checkExists().usingWatcher(this).forPath(path);
- } catch (Exception e2){
- log.warn("Error resetting watch for {}: {}",
- controllerId, e2.getMessage());
- }
- }
- }
- };
+ ServiceInstance<ControllerService> thisInstance = ServiceInstance.<ControllerService>builder()
+ .name(CONTROLLER_SERVICE_NAME)
+ .payload(new ControllerService(controllerId))
+ //.port((int)(65535 * Math.random())) // in a real application, you'd use a common port
+ //.uriSpec(uriSpec)
+ .build();
- //Create ephemeral node in controller registry
- //TODO Use protection
- client.create().withMode(CreateMode.EPHEMERAL)
- .forPath(path, bytes);
- client.checkExists().usingWatcher(watcher).forPath(path);
+ serviceDiscovery.registerService(thisInstance);
} catch (Exception e) {
- throw new RegistryException("Error contacting the Zookeeper service", e);
+ // TODO Auto-generated catch block
+ e.printStackTrace();
}
+
}
@Override
@@ -446,12 +408,22 @@
client = client.usingNamespace(namespace);
- controllerCache = new PathChildrenCache(client, controllerPath, true);
switchCache = new PathChildrenCache(client, switchLatchesPath, true);
switchCache.getListenable().addListener(switchPathCacheListener);
+ //Build the service discovery object
+ serviceDiscovery = ServiceDiscoveryBuilder.builder(ControllerService.class)
+ .client(client).basePath(SERVICES_PATH).build();
+
+ //We read the list of services very frequently (GUI periodically queries them)
+ //so we'll cache them to cut down on Zookeeper queries.
+ serviceCache = serviceDiscovery.serviceCacheBuilder()
+ .name(CONTROLLER_SERVICE_NAME).build();
+
+
try {
- controllerCache.start(StartMode.BUILD_INITIAL_CACHE);
+ serviceDiscovery.start();
+ serviceCache.start();
//Don't prime the cache, we want a notification for each child node in the path
switchCache.start(StartMode.NORMAL);