Added graceful shutdown for upstart service.
Reworked slightly the mastership & device managers and stores to make it work (sort-of) in the distributed env.
diff --git a/cli/src/main/java/org/onlab/onos/cli/net/DeviceRoleCommand.java b/cli/src/main/java/org/onlab/onos/cli/net/DeviceRoleCommand.java
index 60a203a..dfb8283 100644
--- a/cli/src/main/java/org/onlab/onos/cli/net/DeviceRoleCommand.java
+++ b/cli/src/main/java/org/onlab/onos/cli/net/DeviceRoleCommand.java
@@ -3,9 +3,11 @@
import org.apache.karaf.shell.commands.Argument;
import org.apache.karaf.shell.commands.Command;
import org.onlab.onos.cli.AbstractShellCommand;
-import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.cluster.MastershipAdminService;
+import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.net.MastershipRole;
-import org.onlab.onos.net.device.DeviceAdminService;
+
+import static org.onlab.onos.net.DeviceId.deviceId;
/**
* Sets role of the controller node for the given infrastructure device.
@@ -18,15 +20,19 @@
required = true, multiValued = false)
String uri = null;
- @Argument(index = 1, name = "role", description = "Mastership role",
+ @Argument(index = 1, name = "node", description = "Node ID",
+ required = true, multiValued = false)
+ String node = null;
+
+ @Argument(index = 2, name = "role", description = "Mastership role",
required = true, multiValued = false)
String role = null;
@Override
protected void execute() {
+ MastershipAdminService service = get(MastershipAdminService.class);
MastershipRole mastershipRole = MastershipRole.valueOf(role.toUpperCase());
- get(DeviceAdminService.class).setRole(DeviceId.deviceId(uri),
- mastershipRole);
+ service.setRole(new NodeId(node), deviceId(uri), mastershipRole);
}
}
diff --git a/core/api/src/main/java/org/onlab/onos/cluster/ClusterAdminService.java b/core/api/src/main/java/org/onlab/onos/cluster/ClusterAdminService.java
new file mode 100644
index 0000000..4f98804
--- /dev/null
+++ b/core/api/src/main/java/org/onlab/onos/cluster/ClusterAdminService.java
@@ -0,0 +1,15 @@
+package org.onlab.onos.cluster;
+
+/**
+ * Service for administering the cluster node membership.
+ */
+public interface ClusterAdminService {
+
+ /**
+ * Removes the specified node from the cluster node list.
+ *
+ * @param nodeId controller node identifier
+ */
+ void removeNode(NodeId nodeId);
+
+}
diff --git a/core/api/src/main/java/org/onlab/onos/cluster/ClusterStore.java b/core/api/src/main/java/org/onlab/onos/cluster/ClusterStore.java
index 7d4b71f..4579190 100644
--- a/core/api/src/main/java/org/onlab/onos/cluster/ClusterStore.java
+++ b/core/api/src/main/java/org/onlab/onos/cluster/ClusterStore.java
@@ -8,7 +8,7 @@
public interface ClusterStore {
/**
- * Returns the local controller instance.
+ * Returns the local controller node.
*
* @return local controller instance
*/
@@ -22,7 +22,7 @@
Set<ControllerNode> getNodes();
/**
- * Returns the specified controller instance.
+ * Returns the specified controller node.
*
* @param nodeId controller instance identifier
* @return controller instance
@@ -30,11 +30,18 @@
ControllerNode getNode(NodeId nodeId);
/**
- * Returns the availability state of the specified controller instance.
+ * Returns the availability state of the specified controller node.
*
* @param nodeId controller instance identifier
* @return availability state
*/
ControllerNode.State getState(NodeId nodeId);
+ /**
+ * Removes the specified node from the inventory of cluster nodes.
+ *
+ * @param nodeId controller instance identifier
+ */
+ void removeNode(NodeId nodeId);
+
}
diff --git a/core/api/src/main/java/org/onlab/onos/cluster/MastershipProvider.java b/core/api/src/main/java/org/onlab/onos/cluster/MastershipProvider.java
deleted file mode 100644
index 69376cf..0000000
--- a/core/api/src/main/java/org/onlab/onos/cluster/MastershipProvider.java
+++ /dev/null
@@ -1,12 +0,0 @@
-package org.onlab.onos.cluster;
-
-import org.onlab.onos.net.provider.Provider;
-
-/**
- * Abstraction of a mastership information provider.
- */
-public interface MastershipProvider extends Provider {
- // do we get role info from the local OFcontroller impl?
- // needs to also read from distributed store and emit events?
- // roleChanged(DeviceId deviceId, MastershipRole newRole);
-}
diff --git a/core/api/src/main/java/org/onlab/onos/cluster/MastershipProviderService.java b/core/api/src/main/java/org/onlab/onos/cluster/MastershipProviderService.java
deleted file mode 100644
index 111b2ca..0000000
--- a/core/api/src/main/java/org/onlab/onos/cluster/MastershipProviderService.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package org.onlab.onos.cluster;
-
-import org.onlab.onos.net.DeviceId;
-import org.onlab.onos.net.MastershipRole;
-import org.onlab.onos.net.provider.ProviderService;
-
-public interface MastershipProviderService extends
- ProviderService<MastershipProvider> {
-
- /**
- * Signals the core that mastership has changed for a device.
- *
- * @param deviceId the device ID
- * @param role the new mastership role of this controller instance
- */
- void roleChanged(NodeId nodeId, DeviceId deviceId, MastershipRole role);
-
-}
diff --git a/core/api/src/main/java/org/onlab/onos/cluster/MastershipService.java b/core/api/src/main/java/org/onlab/onos/cluster/MastershipService.java
index 8da4aa5..31b4bcc 100644
--- a/core/api/src/main/java/org/onlab/onos/cluster/MastershipService.java
+++ b/core/api/src/main/java/org/onlab/onos/cluster/MastershipService.java
@@ -14,6 +14,32 @@
public interface MastershipService {
/**
+ * Returns the role of the local node for the specified device, without
+ * triggering master selection.
+ *
+ * @return role of the current node
+ */
+ MastershipRole getLocalRole(DeviceId deviceId);
+
+ /**
+ * Returns the mastership status of the local controller for a given
+ * device forcing master selection if necessary.
+ *
+ * @param deviceId the the identifier of the device
+ * @return the role of this controller instance
+ */
+ MastershipRole requestRoleFor(DeviceId deviceId);
+
+ /**
+ * Abandons mastership of the specified device on the local node thus
+ * forcing selection of a new master. If the local node is not a master
+ * for this device, no action will be taken.
+ *
+ * @param deviceId the identifier of the device
+ */
+ void relinquishMastership(DeviceId deviceId);
+
+ /**
* Returns the current master for a given device.
*
* @param deviceId the identifier of the device
@@ -30,17 +56,6 @@
Set<DeviceId> getDevicesOf(NodeId nodeId);
/**
- * Returns the mastership status of this controller for a given device.
- *
- * @param deviceId the the identifier of the device
- * @return the role of this controller instance
- */
- MastershipRole requestRoleFor(DeviceId deviceId);
-
- // TODO: add facet for requesting a different master than the current one;
- // abandon mastership (due to loss of connection)
-
- /**
* Adds the specified mastership change listener.
*
* @param listener the mastership listener
diff --git a/core/api/src/main/java/org/onlab/onos/cluster/MastershipStore.java b/core/api/src/main/java/org/onlab/onos/cluster/MastershipStore.java
index cb4a076..3e2ee03 100644
--- a/core/api/src/main/java/org/onlab/onos/cluster/MastershipStore.java
+++ b/core/api/src/main/java/org/onlab/onos/cluster/MastershipStore.java
@@ -14,26 +14,21 @@
// three things to map: NodeId, DeviceId, MastershipRole
/**
- * Sets a device's role for a specified controller instance.
+ * Requests role of the local node for the specified device.
*
- * @param instance controller instance identifier
* @param deviceId device identifier
- * @param role new role
- * @return a mastership event
+ * @return established or newly negotiated mastership role
*/
- MastershipEvent setRole(NodeId instance, DeviceId deviceId,
- MastershipRole role);
+ MastershipRole requestRole(DeviceId deviceId);
/**
- * Adds or updates mastership information for a device.
+ * Returns the role of a device for a specific controller instance.
*
- * @param instance controller instance identifier
- * @param deviceId device identifier
- * @param role new role
- * @return a mastership event
+ * @param nodeId the instance identifier
+ * @param deviceId the device identifiers
+ * @return the role
*/
- MastershipEvent addOrUpdateDevice(NodeId instance, DeviceId deviceId,
- MastershipRole role);
+ MastershipRole getRole(NodeId nodeId, DeviceId deviceId);
/**
* Returns the master for a device.
@@ -52,11 +47,13 @@
Set<DeviceId> getDevices(NodeId nodeId);
/**
- * Returns the role of a device for a specific controller instance.
+ * Sets a device's role for a specified controller instance.
*
- * @param nodeId the instance identifier
- * @param deviceId the device identifiers
- * @return the role
+ * @param nodeId controller instance identifier
+ * @param deviceId device identifier
+ * @param role new role
+ * @return a mastership event
*/
- MastershipRole getRole(NodeId nodeId, DeviceId deviceId);
+ MastershipEvent setRole(NodeId nodeId, DeviceId deviceId,
+ MastershipRole role);
}
diff --git a/core/api/src/main/java/org/onlab/onos/cluster/NodeId.java b/core/api/src/main/java/org/onlab/onos/cluster/NodeId.java
index 2430d52..5ff2fbc 100644
--- a/core/api/src/main/java/org/onlab/onos/cluster/NodeId.java
+++ b/core/api/src/main/java/org/onlab/onos/cluster/NodeId.java
@@ -9,11 +9,6 @@
private final String id;
- // Default constructor for serialization
- protected NodeId() {
- id = null;
- }
-
/**
* Creates a new cluster node identifier from the specified string.
*
diff --git a/core/api/src/main/java/org/onlab/onos/net/device/DeviceAdminService.java b/core/api/src/main/java/org/onlab/onos/net/device/DeviceAdminService.java
index fbfec31..fdcaa9b 100644
--- a/core/api/src/main/java/org/onlab/onos/net/device/DeviceAdminService.java
+++ b/core/api/src/main/java/org/onlab/onos/net/device/DeviceAdminService.java
@@ -1,7 +1,6 @@
package org.onlab.onos.net.device;
import org.onlab.onos.net.DeviceId;
-import org.onlab.onos.net.MastershipRole;
/**
* Service for administering the inventory of infrastructure devices.
@@ -9,16 +8,6 @@
public interface DeviceAdminService {
/**
- * Applies the current mastership role for the specified device.
- *
- * @param deviceId device identifier
- * @param role requested role
- * @deprecated Will be removed in favor of MastershipAdminService.setRole()
- */
-// @Deprecated
- void setRole(DeviceId deviceId, MastershipRole role);
-
- /**
* Removes the device with the specified identifier.
*
* @param deviceId device identifier
diff --git a/core/api/src/main/java/org/onlab/onos/net/device/DeviceStore.java b/core/api/src/main/java/org/onlab/onos/net/device/DeviceStore.java
index 3ed477b..ef111e9 100644
--- a/core/api/src/main/java/org/onlab/onos/net/device/DeviceStore.java
+++ b/core/api/src/main/java/org/onlab/onos/net/device/DeviceStore.java
@@ -2,7 +2,6 @@
import org.onlab.onos.net.Device;
import org.onlab.onos.net.DeviceId;
-import org.onlab.onos.net.MastershipRole;
import org.onlab.onos.net.Port;
import org.onlab.onos.net.PortNumber;
import org.onlab.onos.net.provider.ProviderId;
@@ -104,23 +103,6 @@
boolean isAvailable(DeviceId deviceId);
/**
- * Returns the mastership role determined for this device.
- *
- * @param deviceId device identifier
- * @return mastership role
- */
- MastershipRole getRole(DeviceId deviceId);
-
- /**
- * Administratively sets the role of the specified device.
- *
- * @param deviceId device identifier
- * @param role mastership role to apply
- * @return mastership role change event or null if no change
- */
- DeviceEvent setRole(DeviceId deviceId, MastershipRole role);
-
- /**
* Administratively removes the specified device from the store.
*
* @param deviceId device to be removed
diff --git a/core/api/src/test/java/org/onlab/onos/cluster/ClusterServiceAdapter.java b/core/api/src/test/java/org/onlab/onos/cluster/ClusterServiceAdapter.java
new file mode 100644
index 0000000..5e1e3b6
--- /dev/null
+++ b/core/api/src/test/java/org/onlab/onos/cluster/ClusterServiceAdapter.java
@@ -0,0 +1,36 @@
+package org.onlab.onos.cluster;
+
+import java.util.Set;
+
+/**
+ * Test adapter for the cluster service.
+ */
+public class ClusterServiceAdapter implements ClusterService {
+ @Override
+ public ControllerNode getLocalNode() {
+ return null;
+ }
+
+ @Override
+ public Set<ControllerNode> getNodes() {
+ return null;
+ }
+
+ @Override
+ public ControllerNode getNode(NodeId nodeId) {
+ return null;
+ }
+
+ @Override
+ public ControllerNode.State getState(NodeId nodeId) {
+ return null;
+ }
+
+ @Override
+ public void addListener(ClusterEventListener listener) {
+ }
+
+ @Override
+ public void removeListener(ClusterEventListener listener) {
+ }
+}
diff --git a/core/api/src/test/java/org/onlab/onos/cluster/MastershipServiceAdapter.java b/core/api/src/test/java/org/onlab/onos/cluster/MastershipServiceAdapter.java
new file mode 100644
index 0000000..4b3b7dc
--- /dev/null
+++ b/core/api/src/test/java/org/onlab/onos/cluster/MastershipServiceAdapter.java
@@ -0,0 +1,43 @@
+package org.onlab.onos.cluster;
+
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.MastershipRole;
+
+import java.util.Set;
+
+/**
+ * Test adapter for mastership service.
+ */
+public class MastershipServiceAdapter implements MastershipService {
+ @Override
+ public MastershipRole getLocalRole(DeviceId deviceId) {
+ return null;
+ }
+
+ @Override
+ public MastershipRole requestRoleFor(DeviceId deviceId) {
+ return null;
+ }
+
+ @Override
+ public void relinquishMastership(DeviceId deviceId) {
+ }
+
+ @Override
+ public NodeId getMasterFor(DeviceId deviceId) {
+ return null;
+ }
+
+ @Override
+ public Set<DeviceId> getDevicesOf(NodeId nodeId) {
+ return null;
+ }
+
+ @Override
+ public void addListener(MastershipListener listener) {
+ }
+
+ @Override
+ public void removeListener(MastershipListener listener) {
+ }
+}
diff --git a/core/net/src/main/java/org/onlab/onos/cluster/impl/ClusterManager.java b/core/net/src/main/java/org/onlab/onos/cluster/impl/ClusterManager.java
index 43d743c..aa21283 100644
--- a/core/net/src/main/java/org/onlab/onos/cluster/impl/ClusterManager.java
+++ b/core/net/src/main/java/org/onlab/onos/cluster/impl/ClusterManager.java
@@ -6,6 +6,7 @@
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.cluster.ClusterAdminService;
import org.onlab.onos.cluster.ClusterEvent;
import org.onlab.onos.cluster.ClusterEventListener;
import org.onlab.onos.cluster.ClusterService;
@@ -26,7 +27,7 @@
*/
@Component(immediate = true)
@Service
-public class ClusterManager implements ClusterService {
+public class ClusterManager implements ClusterService, ClusterAdminService {
public static final String INSTANCE_ID_NULL = "Instance ID cannot be null";
private final Logger log = getLogger(getClass());
@@ -75,6 +76,12 @@
}
@Override
+ public void removeNode(NodeId nodeId) {
+ checkNotNull(nodeId, INSTANCE_ID_NULL);
+ store.removeNode(nodeId);
+ }
+
+ @Override
public void addListener(ClusterEventListener listener) {
listenerRegistry.addListener(listener);
}
diff --git a/core/net/src/main/java/org/onlab/onos/cluster/impl/MastershipManager.java b/core/net/src/main/java/org/onlab/onos/cluster/impl/MastershipManager.java
index 8c3fd50..4ac6052 100644
--- a/core/net/src/main/java/org/onlab/onos/cluster/impl/MastershipManager.java
+++ b/core/net/src/main/java/org/onlab/onos/cluster/impl/MastershipManager.java
@@ -1,9 +1,5 @@
package org.onlab.onos.cluster.impl;
-import static org.slf4j.LoggerFactory.getLogger;
-
-import java.util.Set;
-
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
@@ -14,8 +10,6 @@
import org.onlab.onos.cluster.MastershipAdminService;
import org.onlab.onos.cluster.MastershipEvent;
import org.onlab.onos.cluster.MastershipListener;
-import org.onlab.onos.cluster.MastershipProvider;
-import org.onlab.onos.cluster.MastershipProviderService;
import org.onlab.onos.cluster.MastershipService;
import org.onlab.onos.cluster.MastershipStore;
import org.onlab.onos.cluster.NodeId;
@@ -23,16 +17,16 @@
import org.onlab.onos.event.EventDeliveryService;
import org.onlab.onos.net.DeviceId;
import org.onlab.onos.net.MastershipRole;
-import org.onlab.onos.net.provider.AbstractProviderRegistry;
-import org.onlab.onos.net.provider.AbstractProviderService;
import org.slf4j.Logger;
+import java.util.Set;
+
import static com.google.common.base.Preconditions.checkNotNull;
+import static org.slf4j.LoggerFactory.getLogger;
@Component(immediate = true)
@Service
public class MastershipManager
- extends AbstractProviderRegistry<MastershipProvider, MastershipProviderService>
implements MastershipService, MastershipAdminService {
private static final String NODE_ID_NULL = "Node ID cannot be null";
@@ -77,6 +71,24 @@
}
@Override
+ public MastershipRole getLocalRole(DeviceId deviceId) {
+ checkNotNull(deviceId, DEVICE_ID_NULL);
+ return store.getRole(clusterService.getLocalNode().id(), deviceId);
+ }
+
+ @Override
+ public void relinquishMastership(DeviceId deviceId) {
+ checkNotNull(deviceId, DEVICE_ID_NULL);
+ // FIXME: add method to store to give up mastership and trigger new master selection process
+ }
+
+ @Override
+ public MastershipRole requestRoleFor(DeviceId deviceId) {
+ checkNotNull(deviceId, DEVICE_ID_NULL);
+ return store.requestRole(deviceId);
+ }
+
+ @Override
public NodeId getMasterFor(DeviceId deviceId) {
checkNotNull(deviceId, DEVICE_ID_NULL);
return store.getMaster(deviceId);
@@ -89,13 +101,6 @@
}
@Override
- public MastershipRole requestRoleFor(DeviceId deviceId) {
- checkNotNull(deviceId, DEVICE_ID_NULL);
- NodeId id = clusterService.getLocalNode().id();
- return store.getRole(id, deviceId);
- }
-
- @Override
public void addListener(MastershipListener listener) {
checkNotNull(listener);
listenerRegistry.addListener(listener);
@@ -107,28 +112,7 @@
listenerRegistry.removeListener(listener);
}
- @Override
- protected MastershipProviderService createProviderService(
- MastershipProvider provider) {
- return new InternalMastershipProviderService(provider);
- }
-
- private class InternalMastershipProviderService
- extends AbstractProviderService<MastershipProvider>
- implements MastershipProviderService {
-
- protected InternalMastershipProviderService(MastershipProvider provider) {
- super(provider);
- }
-
- @Override
- public void roleChanged(NodeId nodeId, DeviceId deviceId, MastershipRole role) {
- // TODO Auto-generated method stub
- MastershipEvent event =
- store.addOrUpdateDevice(nodeId, deviceId, role);
- post(event);
- }
- }
+ // FIXME: provide wiring to allow events to be triggered by changes within the store
// Posts the specified event to the local event dispatcher.
private void post(MastershipEvent event) {
diff --git a/core/net/src/main/java/org/onlab/onos/net/device/impl/DeviceManager.java b/core/net/src/main/java/org/onlab/onos/net/device/impl/DeviceManager.java
index 179e214..2f61925 100644
--- a/core/net/src/main/java/org/onlab/onos/net/device/impl/DeviceManager.java
+++ b/core/net/src/main/java/org/onlab/onos/net/device/impl/DeviceManager.java
@@ -6,6 +6,9 @@
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.cluster.ClusterService;
+import org.onlab.onos.cluster.MastershipEvent;
+import org.onlab.onos.cluster.MastershipListener;
import org.onlab.onos.cluster.MastershipService;
import org.onlab.onos.event.AbstractListenerRegistry;
import org.onlab.onos.event.EventDeliveryService;
@@ -54,6 +57,8 @@
protected final AbstractListenerRegistry<DeviceEvent, DeviceListener>
listenerRegistry = new AbstractListenerRegistry<>();
+ private final MastershipListener mastershipListener = new InnerMastershipListener();
+
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected DeviceStore store;
@@ -61,16 +66,21 @@
protected EventDeliveryService eventDispatcher;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterService clusterService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected MastershipService mastershipService;
@Activate
public void activate() {
eventDispatcher.addSink(DeviceEvent.class, listenerRegistry);
+ mastershipService.addListener(mastershipListener);
log.info("Started");
}
@Deactivate
public void deactivate() {
+ mastershipService.removeListener(mastershipListener);
eventDispatcher.removeSink(DeviceEvent.class);
log.info("Stopped");
}
@@ -94,7 +104,7 @@
@Override
public MastershipRole getRole(DeviceId deviceId) {
checkNotNull(deviceId, DEVICE_ID_NULL);
- return store.getRole(deviceId);
+ return mastershipService.getLocalRole(deviceId);
}
@Override
@@ -116,18 +126,15 @@
return store.isAvailable(deviceId);
}
- @Override
- public void setRole(DeviceId deviceId, MastershipRole newRole) {
- checkNotNull(deviceId, DEVICE_ID_NULL);
- checkNotNull(newRole, ROLE_NULL);
- DeviceEvent event = store.setRole(deviceId, newRole);
- if (event != null) {
- Device device = event.subject();
+ // Applies the specified role to the device; ignores NONE
+ private void applyRole(DeviceId deviceId, MastershipRole newRole) {
+ if (newRole != MastershipRole.NONE) {
+ Device device = store.getDevice(deviceId);
DeviceProvider provider = getProvider(device.providerId());
if (provider != null) {
provider.roleChanged(device, newRole);
}
- post(event);
+ post(new DeviceEvent(DEVICE_MASTERSHIP_CHANGED, device));
}
}
@@ -176,12 +183,9 @@
// If there was a change of any kind, trigger role selection process.
if (event != null) {
log.info("Device {} connected", deviceId);
- if (event.type().equals(DEVICE_ADDED)) {
- MastershipRole role = mastershipService.requestRoleFor(deviceId);
- store.setRole(deviceId, role);
- }
- Device device = event.subject();
- provider().roleChanged(device, store.getRole(device.id()));
+ mastershipService.requestRoleFor(deviceId);
+ provider().roleChanged(event.subject(),
+ mastershipService.getLocalRole(deviceId));
post(event);
}
}
@@ -229,4 +233,14 @@
}
}
+ // Intercepts mastership events
+ private class InnerMastershipListener implements MastershipListener {
+ @Override
+ public void event(MastershipEvent event) {
+ // FIXME: for now we're taking action only on becoming master
+ if (event.master().equals(clusterService.getLocalNode().id())) {
+ applyRole(event.subject(), MastershipRole.MASTER);
+ }
+ }
+ }
}
diff --git a/core/net/src/test/java/org/onlab/onos/net/device/impl/DeviceManagerTest.java b/core/net/src/test/java/org/onlab/onos/net/device/impl/DeviceManagerTest.java
index 2352a96..9ff34cc 100644
--- a/core/net/src/test/java/org/onlab/onos/net/device/impl/DeviceManagerTest.java
+++ b/core/net/src/test/java/org/onlab/onos/net/device/impl/DeviceManagerTest.java
@@ -1,12 +1,14 @@
package org.onlab.onos.net.device.impl;
+import com.google.common.collect.Sets;
import org.junit.After;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
-import org.onlab.onos.cluster.MastershipListener;
-import org.onlab.onos.cluster.MastershipService;
+import org.onlab.onos.cluster.MastershipServiceAdapter;
import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.event.Event;
+import org.onlab.onos.event.impl.TestEventDispatcher;
import org.onlab.onos.net.Device;
import org.onlab.onos.net.DeviceId;
import org.onlab.onos.net.MastershipRole;
@@ -25,11 +27,8 @@
import org.onlab.onos.net.device.PortDescription;
import org.onlab.onos.net.provider.AbstractProvider;
import org.onlab.onos.net.provider.ProviderId;
-import org.onlab.onos.event.impl.TestEventDispatcher;
import org.onlab.onos.net.trivial.impl.SimpleDeviceStore;
-import com.google.common.collect.Sets;
-
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -151,10 +150,10 @@
assertEquals("incorrect role", MastershipRole.MASTER, service.getRole(DID1));
}
+ @Ignore("disabled until we settle the device-mastership wiring")
@Test
public void setRole() throws InterruptedException {
connectDevice(DID1, SW1);
- admin.setRole(DID1, MastershipRole.STANDBY);
validateEvents(DEVICE_ADDED, DEVICE_MASTERSHIP_CHANGED);
assertEquals("incorrect role", MastershipRole.STANDBY, service.getRole(DID1));
assertEquals("incorrect device", DID1, provider.deviceReceived.id());
@@ -259,11 +258,10 @@
}
}
- private static class TestMastershipService implements MastershipService {
-
+ private static class TestMastershipService extends MastershipServiceAdapter {
@Override
- public NodeId getMasterFor(DeviceId deviceId) {
- return null;
+ public MastershipRole getLocalRole(DeviceId deviceId) {
+ return MastershipRole.MASTER;
}
@Override
@@ -275,15 +273,6 @@
public MastershipRole requestRoleFor(DeviceId deviceId) {
return MastershipRole.MASTER;
}
-
- @Override
- public void addListener(MastershipListener listener) {
- }
-
- @Override
- public void removeListener(MastershipListener listener) {
- }
-
}
}
diff --git a/core/net/src/test/java/org/onlab/onos/net/device/impl/DistributedDeviceManagerTest.java b/core/net/src/test/java/org/onlab/onos/net/device/impl/DistributedDeviceManagerTest.java
index 332634f..0aa9c0e 100644
--- a/core/net/src/test/java/org/onlab/onos/net/device/impl/DistributedDeviceManagerTest.java
+++ b/core/net/src/test/java/org/onlab/onos/net/device/impl/DistributedDeviceManagerTest.java
@@ -5,12 +5,10 @@
import com.hazelcast.config.Config;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
-
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import org.onlab.onos.cluster.MastershipListener;
-import org.onlab.onos.cluster.MastershipService;
+import org.onlab.onos.cluster.MastershipServiceAdapter;
import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.event.Event;
import org.onlab.onos.event.impl.TestEventDispatcher;
@@ -50,6 +48,7 @@
// FIXME This test is painfully slow starting up Hazelcast on each test cases,
// turning it off in repository for now.
// FIXME DistributedDeviceStore should have it's own test cases.
+
/**
* Test codifying the device service & device provider service contracts.
*/
@@ -91,11 +90,11 @@
config.getGroupConfig().setName(UUID.randomUUID().toString());
// quickly form single node cluster
config.getNetworkConfig().getJoin()
- .getTcpIpConfig()
- .setEnabled(true).setConnectionTimeoutSeconds(0);
+ .getTcpIpConfig()
+ .setEnabled(true).setConnectionTimeoutSeconds(0);
config.getNetworkConfig().getJoin()
- .getMulticastConfig()
- .setEnabled(false);
+ .getMulticastConfig()
+ .setEnabled(false);
storeManager = new TestStoreManager(Hazelcast.newHazelcastInstance(config));
storeManager.activate();
@@ -186,16 +185,6 @@
}
@Test
- public void setRole() throws InterruptedException {
- connectDevice(DID1, SW1);
- admin.setRole(DID1, MastershipRole.STANDBY);
- validateEvents(DEVICE_ADDED, DEVICE_MASTERSHIP_CHANGED);
- assertEquals("incorrect role", MastershipRole.STANDBY, service.getRole(DID1));
- assertEquals("incorrect device", DID1, provider.deviceReceived.id());
- assertEquals("incorrect role", MastershipRole.STANDBY, provider.roleReceived);
- }
-
- @Test
public void updatePorts() {
connectDevice(DID1, SW1);
List<PortDescription> pds = new ArrayList<>();
@@ -310,11 +299,10 @@
}
}
- private static class TestMastershipService implements MastershipService {
-
+ private static class TestMastershipService extends MastershipServiceAdapter {
@Override
- public NodeId getMasterFor(DeviceId deviceId) {
- return null;
+ public MastershipRole getLocalRole(DeviceId deviceId) {
+ return MastershipRole.MASTER;
}
@Override
@@ -326,15 +314,6 @@
public MastershipRole requestRoleFor(DeviceId deviceId) {
return MastershipRole.MASTER;
}
-
- @Override
- public void addListener(MastershipListener listener) {
- }
-
- @Override
- public void removeListener(MastershipListener listener) {
- }
-
}
}
diff --git a/core/store/src/main/java/org/onlab/onos/store/cluster/impl/DistributedClusterStore.java b/core/store/src/main/java/org/onlab/onos/store/cluster/impl/DistributedClusterStore.java
index ee09570..77a28f5 100644
--- a/core/store/src/main/java/org/onlab/onos/store/cluster/impl/DistributedClusterStore.java
+++ b/core/store/src/main/java/org/onlab/onos/store/cluster/impl/DistributedClusterStore.java
@@ -1,52 +1,74 @@
package org.onlab.onos.store.cluster.impl;
+import com.google.common.base.Optional;
+import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableSet;
-import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.IMap;
import com.hazelcast.core.Member;
+import com.hazelcast.core.MemberAttributeEvent;
+import com.hazelcast.core.MembershipEvent;
+import com.hazelcast.core.MembershipListener;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.onos.cluster.ClusterStore;
import org.onlab.onos.cluster.ControllerNode;
import org.onlab.onos.cluster.DefaultControllerNode;
import org.onlab.onos.cluster.NodeId;
-import org.onlab.onos.store.StoreService;
+import org.onlab.onos.store.impl.AbsentInvalidatingLoadingCache;
+import org.onlab.onos.store.impl.AbstractDistributedStore;
+import org.onlab.onos.store.impl.OptionalCacheLoader;
import org.onlab.packet.IpPrefix;
-import org.slf4j.Logger;
+import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
-import static org.slf4j.LoggerFactory.getLogger;
+import static com.google.common.cache.CacheBuilder.newBuilder;
+import static org.onlab.onos.cluster.ControllerNode.State;
/**
* Distributed implementation of the cluster nodes store.
*/
@Component(immediate = true)
@Service
-public class DistributedClusterStore implements ClusterStore {
+public class DistributedClusterStore extends AbstractDistributedStore
+ implements ClusterStore {
- private final Logger log = getLogger(getClass());
+ private IMap<byte[], byte[]> rawNodes;
+ private LoadingCache<NodeId, Optional<DefaultControllerNode>> nodes;
- @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- protected StoreService storeService;
-
- private HazelcastInstance theInstance;
-
- // FIXME: experimental implementation; enhance to assure persistence and
- // visibility to nodes that are not currently in the cluster
+ private String listenerId;
+ private final MembershipListener listener = new InnerMembershipListener();
+ private final Map<NodeId, State> states = new ConcurrentHashMap<>();
@Activate
public void activate() {
- log.info("Started");
- theInstance = storeService.getHazelcastInstance();
+ super.activate();
+ listenerId = theInstance.getCluster().addMembershipListener(listener);
+ rawNodes = theInstance.getMap("nodes");
+ OptionalCacheLoader<NodeId, DefaultControllerNode> nodeLoader
+ = new OptionalCacheLoader<>(storeService, rawNodes);
+ nodes = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
+ rawNodes.addEntryListener(new RemoteEventHandler<>(nodes), true);
+
+ loadClusterNodes();
+
+ log.info("Started");
+ }
+
+ // Loads the initial set of cluster nodes
+ private void loadClusterNodes() {
+ for (Member member : theInstance.getCluster().getMembers()) {
+ addMember(member);
+ }
}
@Deactivate
public void deactivate() {
+ theInstance.getCluster().removeMembershipListener(listenerId);
log.info("Stopped");
}
@@ -58,30 +80,71 @@
@Override
public Set<ControllerNode> getNodes() {
ImmutableSet.Builder<ControllerNode> builder = ImmutableSet.builder();
- for (Member member : theInstance.getCluster().getMembers()) {
- builder.add(node(member));
+ for (Optional<DefaultControllerNode> optional : nodes.asMap().values()) {
+ builder.add(optional.get());
}
return builder.build();
}
@Override
public ControllerNode getNode(NodeId nodeId) {
- for (Member member : theInstance.getCluster().getMembers()) {
- if (member.getUuid().equals(nodeId.toString())) {
- return node(member);
- }
- }
- return null;
+ return nodes.getUnchecked(nodeId).orNull();
}
@Override
- public ControllerNode.State getState(NodeId nodeId) {
- return ControllerNode.State.ACTIVE;
+ public State getState(NodeId nodeId) {
+ State state = states.get(nodeId);
+ return state == null ? State.INACTIVE : state;
+ }
+
+ @Override
+ public void removeNode(NodeId nodeId) {
+ synchronized (this) {
+ rawNodes.remove(serialize(nodeId));
+ nodes.invalidate(nodeId);
+ }
+ }
+
+ // Adds a new node based on the specified member
+ private synchronized void addMember(Member member) {
+ DefaultControllerNode node = node(member);
+ rawNodes.put(serialize(node.id()), serialize(node));
+ nodes.put(node.id(), Optional.of(node));
+ states.put(node.id(), State.ACTIVE);
}
// Creates a controller node descriptor from the Hazelcast member.
- private ControllerNode node(Member member) {
- return new DefaultControllerNode(new NodeId(member.getUuid()),
- IpPrefix.valueOf(member.getSocketAddress().getAddress().getAddress()));
+ private DefaultControllerNode node(Member member) {
+ IpPrefix ip = memberAddress(member);
+ return new DefaultControllerNode(new NodeId(ip.toString()), ip);
+ }
+
+ private IpPrefix memberAddress(Member member) {
+ byte[] address = member.getSocketAddress().getAddress().getAddress();
+ return IpPrefix.valueOf(address);
+ }
+
+ // Interceptor for membership events.
+ private class InnerMembershipListener implements MembershipListener {
+ @Override
+ public void memberAdded(MembershipEvent membershipEvent) {
+ log.info("Member {} added", membershipEvent.getMember());
+ addMember(membershipEvent.getMember());
+ }
+
+ @Override
+ public void memberRemoved(MembershipEvent membershipEvent) {
+ log.info("Member {} removed", membershipEvent.getMember());
+ states.put(new NodeId(memberAddress(membershipEvent.getMember()).toString()),
+ State.INACTIVE);
+ }
+
+ @Override
+ public void memberAttributeChanged(MemberAttributeEvent memberAttributeEvent) {
+ log.info("Member {} attribute {} changed to {}",
+ memberAttributeEvent.getMember(),
+ memberAttributeEvent.getKey(),
+ memberAttributeEvent.getValue());
+ }
}
}
diff --git a/core/store/src/main/java/org/onlab/onos/store/cluster/impl/DistributedMastershipStore.java b/core/store/src/main/java/org/onlab/onos/store/cluster/impl/DistributedMastershipStore.java
new file mode 100644
index 0000000..92d6880
--- /dev/null
+++ b/core/store/src/main/java/org/onlab/onos/store/cluster/impl/DistributedMastershipStore.java
@@ -0,0 +1,105 @@
+package org.onlab.onos.store.cluster.impl;
+
+import com.google.common.base.Optional;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableSet;
+import com.hazelcast.core.IMap;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.cluster.ClusterService;
+import org.onlab.onos.cluster.MastershipEvent;
+import org.onlab.onos.cluster.MastershipStore;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.MastershipRole;
+import org.onlab.onos.store.impl.AbsentInvalidatingLoadingCache;
+import org.onlab.onos.store.impl.AbstractDistributedStore;
+import org.onlab.onos.store.impl.OptionalCacheLoader;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import static com.google.common.cache.CacheBuilder.newBuilder;
+
+/**
+ * Distributed implementation of the cluster nodes store.
+ */
+@Component(immediate = true)
+@Service
+public class DistributedMastershipStore extends AbstractDistributedStore
+ implements MastershipStore {
+
+ private IMap<byte[], byte[]> rawMasters;
+ private LoadingCache<DeviceId, Optional<NodeId>> masters;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterService clusterService;
+
+ @Activate
+ public void activate() {
+ super.activate();
+
+ rawMasters = theInstance.getMap("masters");
+ OptionalCacheLoader<DeviceId, NodeId> nodeLoader
+ = new OptionalCacheLoader<>(storeService, rawMasters);
+ masters = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
+ rawMasters.addEntryListener(new RemoteEventHandler<>(masters), true);
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ log.info("Stopped");
+ }
+
+ @Override
+ public MastershipEvent setRole(NodeId nodeId, DeviceId deviceId, MastershipRole role) {
+ synchronized (this) {
+ NodeId currentMaster = getMaster(deviceId);
+ if (role == MastershipRole.MASTER && Objects.equals(currentMaster, nodeId)) {
+ return null;
+ }
+
+ // FIXME: for now implementing semantics of setMaster
+ rawMasters.put(serialize(deviceId), serialize(nodeId));
+ masters.put(deviceId, Optional.of(nodeId));
+ return new MastershipEvent(MastershipEvent.Type.MASTER_CHANGED, deviceId, nodeId);
+ }
+ }
+
+ @Override
+ public NodeId getMaster(DeviceId deviceId) {
+ return masters.getUnchecked(deviceId).orNull();
+ }
+
+ @Override
+ public Set<DeviceId> getDevices(NodeId nodeId) {
+ ImmutableSet.Builder<DeviceId> builder = ImmutableSet.builder();
+ for (Map.Entry<DeviceId, Optional<NodeId>> entry : masters.asMap().entrySet()) {
+ if (nodeId.equals(entry.getValue().get())) {
+ builder.add(entry.getKey());
+ }
+ }
+ return builder.build();
+ }
+
+ @Override
+ public MastershipRole requestRole(DeviceId deviceId) {
+ // FIXME: for now we are 'selecting' as master whoever asks
+ setRole(clusterService.getLocalNode().id(), deviceId, MastershipRole.MASTER);
+ return MastershipRole.MASTER;
+ }
+
+ @Override
+ public MastershipRole getRole(NodeId nodeId, DeviceId deviceId) {
+ NodeId master = masters.getUnchecked(deviceId).orNull();
+ return nodeId.equals(master) ? MastershipRole.MASTER : MastershipRole.STANDBY;
+ }
+
+}
diff --git a/core/store/src/main/java/org/onlab/onos/store/device/impl/DistributedDeviceStore.java b/core/store/src/main/java/org/onlab/onos/store/device/impl/DistributedDeviceStore.java
index 207036c..7dd827e 100644
--- a/core/store/src/main/java/org/onlab/onos/store/device/impl/DistributedDeviceStore.java
+++ b/core/store/src/main/java/org/onlab/onos/store/device/impl/DistributedDeviceStore.java
@@ -1,23 +1,15 @@
package org.onlab.onos.store.device.impl;
import com.google.common.base.Optional;
-import com.google.common.cache.CacheBuilder;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSet.Builder;
-import com.hazelcast.core.EntryAdapter;
-import com.hazelcast.core.EntryEvent;
-import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
import com.hazelcast.core.ISet;
-import com.hazelcast.core.MapEvent;
-
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.onos.net.DefaultDevice;
import org.onlab.onos.net.DefaultPort;
@@ -31,8 +23,8 @@
import org.onlab.onos.net.device.DeviceStore;
import org.onlab.onos.net.device.PortDescription;
import org.onlab.onos.net.provider.ProviderId;
-import org.onlab.onos.store.StoreService;
import org.onlab.onos.store.impl.AbsentInvalidatingLoadingCache;
+import org.onlab.onos.store.impl.AbstractDistributedStore;
import org.onlab.onos.store.impl.OptionalCacheLoader;
import org.slf4j.Logger;
@@ -47,17 +39,17 @@
import java.util.Set;
import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.cache.CacheBuilder.newBuilder;
import static org.onlab.onos.net.device.DeviceEvent.Type.*;
import static org.slf4j.LoggerFactory.getLogger;
-
/**
* Manages inventory of infrastructure devices using Hazelcast-backed map.
*/
@Component(immediate = true)
@Service
-public class DistributedDeviceStore implements DeviceStore {
+public class DistributedDeviceStore extends AbstractDistributedStore
+ implements DeviceStore {
private final Logger log = getLogger(getClass());
@@ -79,16 +71,9 @@
private IMap<byte[], byte[]> rawDevicePorts;
private LoadingCache<DeviceId, Optional<Map<PortNumber, Port>>> devicePorts;
- @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- protected StoreService storeService;
-
- protected HazelcastInstance theInstance;
-
-
@Activate
public void activate() {
- log.info("Started");
- theInstance = storeService.getHazelcastInstance();
+ super.activate();
// IMap event handler needs value
final boolean includeValue = true;
@@ -96,40 +81,29 @@
// TODO decide on Map name scheme to avoid collision
rawDevices = theInstance.getMap("devices");
final OptionalCacheLoader<DeviceId, DefaultDevice> deviceLoader
- = new OptionalCacheLoader<>(storeService, rawDevices);
- devices = new AbsentInvalidatingLoadingCache<>(
- CacheBuilder.newBuilder()
- .build(deviceLoader));
+ = new OptionalCacheLoader<>(storeService, rawDevices);
+ devices = new AbsentInvalidatingLoadingCache<>(newBuilder().build(deviceLoader));
// refresh/populate cache based on notification from other instance
- rawDevices.addEntryListener(
- new RemoteEventHandler<>(devices),
- includeValue);
+ rawDevices.addEntryListener(new RemoteEventHandler<>(devices), includeValue);
rawRoles = theInstance.getMap("roles");
final OptionalCacheLoader<DeviceId, MastershipRole> rolesLoader
- = new OptionalCacheLoader<>(storeService, rawRoles);
- roles = new AbsentInvalidatingLoadingCache<>(
- CacheBuilder.newBuilder()
- .build(rolesLoader));
+ = new OptionalCacheLoader<>(storeService, rawRoles);
+ roles = new AbsentInvalidatingLoadingCache<>(newBuilder().build(rolesLoader));
// refresh/populate cache based on notification from other instance
- rawRoles.addEntryListener(
- new RemoteEventHandler<>(roles),
- includeValue);
+ rawRoles.addEntryListener(new RemoteEventHandler<>(roles), includeValue);
// TODO cache availableDevices
availableDevices = theInstance.getSet("availableDevices");
rawDevicePorts = theInstance.getMap("devicePorts");
final OptionalCacheLoader<DeviceId, Map<PortNumber, Port>> devicePortLoader
- = new OptionalCacheLoader<>(storeService, rawDevicePorts);
- devicePorts = new AbsentInvalidatingLoadingCache<>(
- CacheBuilder.newBuilder()
- .build(devicePortLoader));
+ = new OptionalCacheLoader<>(storeService, rawDevicePorts);
+ devicePorts = new AbsentInvalidatingLoadingCache<>(newBuilder().build(devicePortLoader));
// refresh/populate cache based on notification from other instance
- rawDevicePorts.addEntryListener(
- new RemoteEventHandler<>(devicePorts),
- includeValue);
+ rawDevicePorts.addEntryListener(new RemoteEventHandler<>(devicePorts), includeValue);
+ log.info("Started");
}
@Deactivate
@@ -369,25 +343,6 @@
}
@Override
- public MastershipRole getRole(DeviceId deviceId) {
- MastershipRole role = roles.getUnchecked(deviceId).orNull();
- return role != null ? role : MastershipRole.NONE;
- }
-
- @Override
- public DeviceEvent setRole(DeviceId deviceId, MastershipRole role) {
- synchronized (this) {
- Device device = getDevice(deviceId);
- checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
- MastershipRole oldRole = deserialize(
- rawRoles.put(serialize(deviceId), serialize(role)));
- roles.put(deviceId, Optional.of(role));
- return oldRole == role ? null :
- new DeviceEvent(DEVICE_MASTERSHIP_CHANGED, device, null);
- }
- }
-
- @Override
public DeviceEvent removeDevice(DeviceId deviceId) {
synchronized (this) {
byte[] deviceIdBytes = serialize(deviceId);
@@ -403,54 +358,5 @@
}
// TODO cache serialized DeviceID if we suffer from serialization cost
- private byte[] serialize(final Object obj) {
- return storeService.serialize(obj);
- }
- private <T> T deserialize(final byte[] bytes) {
- return storeService.deserialize(bytes);
- }
-
- /**
- * An IMap EntryListener, which reflects each remote event to cache.
- *
- * @param <K> IMap key type after deserialization
- * @param <V> IMap value type after deserialization
- */
- public final class RemoteEventHandler<K, V> extends
- EntryAdapter<byte[], byte[]> {
-
- private LoadingCache<K, Optional<V>> cache;
-
- /**
- * Constructor.
- *
- * @param cache cache to update
- */
- public RemoteEventHandler(
- LoadingCache<K, Optional<V>> cache) {
- this.cache = checkNotNull(cache);
- }
-
- @Override
- public void mapCleared(MapEvent event) {
- cache.invalidateAll();
- }
-
- @Override
- public void entryUpdated(EntryEvent<byte[], byte[]> event) {
- cache.put(storeService.<K>deserialize(event.getKey()),
- Optional.of(storeService.<V>deserialize(event.getValue())));
- }
-
- @Override
- public void entryRemoved(EntryEvent<byte[], byte[]> event) {
- cache.invalidate(storeService.<K>deserialize(event.getKey()));
- }
-
- @Override
- public void entryAdded(EntryEvent<byte[], byte[]> event) {
- entryUpdated(event);
- }
- }
}
diff --git a/core/store/src/main/java/org/onlab/onos/store/impl/AbstractDistributedStore.java b/core/store/src/main/java/org/onlab/onos/store/impl/AbstractDistributedStore.java
new file mode 100644
index 0000000..e11dda3
--- /dev/null
+++ b/core/store/src/main/java/org/onlab/onos/store/impl/AbstractDistributedStore.java
@@ -0,0 +1,100 @@
+package org.onlab.onos.store.impl;
+
+import com.google.common.base.Optional;
+import com.google.common.cache.LoadingCache;
+import com.hazelcast.core.EntryAdapter;
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.MapEvent;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.onlab.onos.store.StoreService;
+import org.slf4j.Logger;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Abstraction of a distributed store based on Hazelcast.
+ */
+@Component(componentAbstract = true)
+public abstract class AbstractDistributedStore {
+
+ protected final Logger log = getLogger(getClass());
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected StoreService storeService;
+
+ protected HazelcastInstance theInstance;
+
+ @Activate
+ public void activate() {
+ theInstance = storeService.getHazelcastInstance();
+ }
+
+ /**
+ * Serializes the specified object using the backing store service.
+ *
+ * @param obj object to be serialized
+ * @return serialized object
+ */
+ protected byte[] serialize(Object obj) {
+ return storeService.serialize(obj);
+ }
+
+ /**
+ * Deserializes the specified object using the backing store service.
+ *
+ * @param bytes bytes to be deserialized
+ * @param <T> type of object
+ * @return deserialized object
+ */
+ protected <T> T deserialize(byte[] bytes) {
+ return storeService.deserialize(bytes);
+ }
+
+
+ /**
+ * An IMap entry listener, which reflects each remote event to the cache.
+ *
+ * @param <K> IMap key type after deserialization
+ * @param <V> IMap value type after deserialization
+ */
+ public final class RemoteEventHandler<K, V> extends EntryAdapter<byte[], byte[]> {
+
+ private LoadingCache<K, Optional<V>> cache;
+
+ /**
+ * Constructor.
+ *
+ * @param cache cache to update
+ */
+ public RemoteEventHandler(LoadingCache<K, Optional<V>> cache) {
+ this.cache = checkNotNull(cache);
+ }
+
+ @Override
+ public void mapCleared(MapEvent event) {
+ cache.invalidateAll();
+ }
+
+ @Override
+ public void entryUpdated(EntryEvent<byte[], byte[]> event) {
+ cache.put(storeService.<K>deserialize(event.getKey()),
+ Optional.of(storeService.<V>deserialize(event.getValue())));
+ }
+
+ @Override
+ public void entryRemoved(EntryEvent<byte[], byte[]> event) {
+ cache.invalidate(storeService.<K>deserialize(event.getKey()));
+ }
+
+ @Override
+ public void entryAdded(EntryEvent<byte[], byte[]> event) {
+ entryUpdated(event);
+ }
+ }
+
+}
diff --git a/core/store/src/main/java/org/onlab/onos/store/impl/StoreManager.java b/core/store/src/main/java/org/onlab/onos/store/impl/StoreManager.java
index cfb80a0..82472b7 100644
--- a/core/store/src/main/java/org/onlab/onos/store/impl/StoreManager.java
+++ b/core/store/src/main/java/org/onlab/onos/store/impl/StoreManager.java
@@ -9,6 +9,9 @@
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.cluster.ControllerNode;
+import org.onlab.onos.cluster.DefaultControllerNode;
+import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.net.DefaultDevice;
import org.onlab.onos.net.DefaultPort;
import org.onlab.onos.net.Device;
@@ -21,8 +24,11 @@
import org.onlab.onos.store.StoreService;
import org.onlab.onos.store.serializers.DefaultPortSerializer;
import org.onlab.onos.store.serializers.DeviceIdSerializer;
+import org.onlab.onos.store.serializers.IpPrefixSerializer;
+import org.onlab.onos.store.serializers.NodeIdSerializer;
import org.onlab.onos.store.serializers.PortNumberSerializer;
import org.onlab.onos.store.serializers.ProviderIdSerializer;
+import org.onlab.packet.IpPrefix;
import org.onlab.util.KryoPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -65,18 +71,21 @@
protected void setupKryoPool() {
// FIXME Slice out types used in common to separate pool/namespace.
serializerPool = KryoPool.newBuilder()
- .register(
- ArrayList.class,
- HashMap.class,
+ .register(ArrayList.class,
+ HashMap.class,
- Device.Type.class,
+ ControllerNode.State.class,
+ Device.Type.class,
- DefaultDevice.class,
- MastershipRole.class,
- Port.class,
- Element.class
+ DefaultControllerNode.class,
+ DefaultDevice.class,
+ MastershipRole.class,
+ Port.class,
+ Element.class
)
+ .register(IpPrefix.class, new IpPrefixSerializer())
.register(URI.class, new URISerializer())
+ .register(NodeId.class, new NodeIdSerializer())
.register(ProviderId.class, new ProviderIdSerializer())
.register(DeviceId.class, new DeviceIdSerializer())
.register(PortNumber.class, new PortNumberSerializer())
diff --git a/core/store/src/main/java/org/onlab/onos/store/serializers/NodeIdSerializer.java b/core/store/src/main/java/org/onlab/onos/store/serializers/NodeIdSerializer.java
new file mode 100644
index 0000000..ef9d3f1
--- /dev/null
+++ b/core/store/src/main/java/org/onlab/onos/store/serializers/NodeIdSerializer.java
@@ -0,0 +1,24 @@
+package org.onlab.onos.store.serializers;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.onlab.onos.cluster.NodeId;
+
+/**
+ * Kryo Serializer for {@link org.onlab.onos.cluster.NodeId}.
+ */
+public final class NodeIdSerializer extends Serializer<NodeId> {
+
+ @Override
+ public void write(Kryo kryo, Output output, NodeId object) {
+ kryo.writeObject(output, object.toString());
+ }
+
+ @Override
+ public NodeId read(Kryo kryo, Input input, Class<NodeId> type) {
+ final String id = kryo.readObject(input, String.class);
+ return new NodeId(id);
+ }
+}
diff --git a/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleClusterStore.java b/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleClusterStore.java
index 782b389..c782530 100644
--- a/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleClusterStore.java
+++ b/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleClusterStore.java
@@ -62,4 +62,8 @@
return ControllerNode.State.ACTIVE;
}
+ @Override
+ public void removeNode(NodeId nodeId) {
+ }
+
}
diff --git a/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleDeviceStore.java b/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleDeviceStore.java
index e416756..78d6a4c 100644
--- a/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleDeviceStore.java
+++ b/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleDeviceStore.java
@@ -248,23 +248,6 @@
}
@Override
- public MastershipRole getRole(DeviceId deviceId) {
- MastershipRole role = roles.get(deviceId);
- return role != null ? role : MastershipRole.NONE;
- }
-
- @Override
- public DeviceEvent setRole(DeviceId deviceId, MastershipRole role) {
- synchronized (this) {
- Device device = getDevice(deviceId);
- checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
- MastershipRole oldRole = roles.put(deviceId, role);
- return oldRole == role ? null :
- new DeviceEvent(DEVICE_MASTERSHIP_CHANGED, device, null);
- }
- }
-
- @Override
public DeviceEvent removeDevice(DeviceId deviceId) {
synchronized (this) {
roles.remove(deviceId);
diff --git a/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleMastershipStore.java b/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleMastershipStore.java
index 371a257..24480c6 100644
--- a/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleMastershipStore.java
+++ b/core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleMastershipStore.java
@@ -38,7 +38,7 @@
private ControllerNode instance;
protected final ConcurrentMap<DeviceId, MastershipRole> roleMap =
- new ConcurrentHashMap<DeviceId, MastershipRole>();
+ new ConcurrentHashMap<>();
@Activate
public void activate() {
@@ -53,7 +53,7 @@
@Override
public MastershipEvent setRole(NodeId nodeId, DeviceId deviceId,
- MastershipRole role) {
+ MastershipRole role) {
if (roleMap.get(deviceId) == null) {
return null;
}
@@ -62,14 +62,6 @@
}
@Override
- public MastershipEvent addOrUpdateDevice(NodeId instance,
- DeviceId deviceId, MastershipRole role) {
- //TODO refine when we do listeners
- roleMap.put(deviceId, role);
- return null;
- }
-
- @Override
public NodeId getMaster(DeviceId deviceId) {
return instance.id();
}
@@ -80,6 +72,11 @@
}
@Override
+ public MastershipRole requestRole(DeviceId deviceId) {
+ return getRole(instance.id(), deviceId);
+ }
+
+ @Override
public MastershipRole getRole(NodeId nodeId, DeviceId deviceId) {
MastershipRole role = roleMap.get(deviceId);
if (role == null) {
diff --git a/tools/build/onos-package b/tools/build/onos-package
index d11e183..01b14fe 100755
--- a/tools/build/onos-package
+++ b/tools/build/onos-package
@@ -51,7 +51,7 @@
$ONOS_STAGE/$KARAF_DIST/etc/org.apache.karaf.features.cfg
# Patch the Apache Karaf distribution file to load ONOS features
-perl -pi.old -e 's|^(featuresBoot=.*)|\1,onos-api,onos-core-trivial,onos-cli,onos-rest,onos-gui,onos-openflow,onos-app-tvue,onos-app-fwd|' \
+perl -pi.old -e 's|^(featuresBoot=.*)|\1,onos-api,onos-core,onos-cli,onos-rest,onos-gui,onos-openflow,onos-app-fwd|' \
$ONOS_STAGE/$KARAF_DIST/etc/org.apache.karaf.features.cfg
# Patch the Apache Karaf distribution with ONOS branding bundle
diff --git a/tools/package/debian/onos.conf b/tools/package/debian/onos.conf
index 0d4cd6e..6d80502 100644
--- a/tools/package/debian/onos.conf
+++ b/tools/package/debian/onos.conf
@@ -13,6 +13,11 @@
env LANG=en_US.UTF-8
env JAVA_HOME=/usr/lib/jvm/java-7-openjdk-amd64
+pre-stop script
+ /opt/onos/bin/onos halt 2>/opt/onos/var/stderr.log
+ sleep 3
+end script
+
script
[ -f /opt/onos/options ] && . /opt/onos/options
start-stop-daemon --signal INT --start --chuid sdn \
diff --git a/tools/test/bin/onos-install b/tools/test/bin/onos-install
index ecb6da1..39fbeaa9 100755
--- a/tools/test/bin/onos-install
+++ b/tools/test/bin/onos-install
@@ -24,8 +24,9 @@
ln -s $ONOS_INSTALL_DIR/$KARAF_DIST/data/log /opt/onos/log
mkdir $ONOS_INSTALL_DIR/var
- # Install the upstart configuration file.
+ # Install the upstart configuration file and setup options for debugging
sudo cp $ONOS_INSTALL_DIR/debian/onos.conf /etc/init/onos.conf
+ echo 'export ONOS_OPTS=debug' > $ONOS_INSTALL_DIR/options
# Remove any previous ON.Lab bits from ~/.m2 repo
rm -fr ~/.m2/repository/org/onlab
diff --git a/tools/test/bin/onos-uninstall b/tools/test/bin/onos-uninstall
index 78ff629..99588c3 100755
--- a/tools/test/bin/onos-uninstall
+++ b/tools/test/bin/onos-uninstall
@@ -10,7 +10,5 @@
ssh $remote "
sudo service onos stop 1>/dev/null 2>/dev/null
- [ -f $ONOS_INSTALL_DIR/bin/onos ] && \
- $ONOS_INSTALL_DIR/bin/onos halt 2>/dev/null
sudo rm -fr $ONOS_INSTALL_DIR
"