In preparation for dynamic clustering support:
- Added Cluster metadata service and metadata store interfaces
- Added a static cluster metadata store implementation that is backed by a local file.
- Consolidated the existing cluster.json and tablets.json metadata files into a single cluster.json file that has all the cluster related metadata.
- Removed dependency on ONOS_NIC env variable.
Change-Id: Ia0a8bb69740caecdcdde71a9408be37c56ae2504
diff --git a/core/net/src/main/java/org/onosproject/cluster/impl/ClusterMetadataManager.java b/core/net/src/main/java/org/onosproject/cluster/impl/ClusterMetadataManager.java
new file mode 100644
index 0000000..a0f7a83
--- /dev/null
+++ b/core/net/src/main/java/org/onosproject/cluster/impl/ClusterMetadataManager.java
@@ -0,0 +1,116 @@
+package org.onosproject.cluster.impl;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.net.SocketException;
+import java.util.Collection;
+import java.util.Enumeration;
+
+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.packet.IpAddress;
+import org.onosproject.cluster.ClusterMetadata;
+import org.onosproject.cluster.ClusterMetadataEvent;
+import org.onosproject.cluster.ClusterMetadataEventListener;
+import org.onosproject.cluster.ClusterMetadataService;
+import org.onosproject.cluster.ClusterMetadataStore;
+import org.onosproject.cluster.ClusterMetadataStoreDelegate;
+import org.onosproject.cluster.ControllerNode;
+import org.onosproject.event.AbstractListenerManager;
+import org.onosproject.store.service.Versioned;
+import org.slf4j.Logger;
+
+/**
+ * Implementation of ClusterMetadataService.
+ */
+@Component(immediate = true)
+@Service
+public class ClusterMetadataManager
+ extends AbstractListenerManager<ClusterMetadataEvent, ClusterMetadataEventListener>
+ implements ClusterMetadataService {
+
+ private ControllerNode localNode;
+ private final Logger log = getLogger(getClass());
+
+ private ClusterMetadataStoreDelegate delegate = new InternalStoreDelegate();
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterMetadataStore store;
+
+ @Activate
+ public void activate() {
+ store.setDelegate(delegate);
+ eventDispatcher.addSink(ClusterMetadataEvent.class, listenerRegistry);
+ establishSelfIdentity();
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ store.unsetDelegate(delegate);
+ eventDispatcher.removeSink(ClusterMetadataEvent.class);
+ log.info("Stopped");
+ }
+
+ @Override
+ public ClusterMetadata getClusterMetadata() {
+ return Versioned.valueOrElse(store.getClusterMetadata(), null);
+ }
+
+ @Override
+ public ControllerNode getLocalNode() {
+ return localNode;
+ }
+
+ @Override
+ public void setClusterMetadata(ClusterMetadata metadata) {
+ checkNotNull(metadata, "Cluster metadata cannot be null");
+ store.setClusterMetadata(metadata);
+ }
+
+ // Store delegate to re-post events emitted from the store.
+ private class InternalStoreDelegate implements ClusterMetadataStoreDelegate {
+ @Override
+ public void notify(ClusterMetadataEvent event) {
+ post(event);
+ }
+ }
+
+ private IpAddress findLocalIp(Collection<ControllerNode> controllerNodes) throws SocketException {
+ Enumeration<NetworkInterface> interfaces =
+ NetworkInterface.getNetworkInterfaces();
+ while (interfaces.hasMoreElements()) {
+ NetworkInterface iface = interfaces.nextElement();
+ Enumeration<InetAddress> inetAddresses = iface.getInetAddresses();
+ while (inetAddresses.hasMoreElements()) {
+ IpAddress ip = IpAddress.valueOf(inetAddresses.nextElement());
+ if (controllerNodes.stream()
+ .map(ControllerNode::ip)
+ .anyMatch(nodeIp -> ip.equals(nodeIp))) {
+ return ip;
+ }
+ }
+ }
+ throw new IllegalStateException("Unable to determine local ip");
+ }
+
+ private void establishSelfIdentity() {
+ try {
+ IpAddress ip = findLocalIp(getClusterMetadata().getNodes());
+ localNode = getClusterMetadata().getNodes()
+ .stream()
+ .filter(node -> node.ip().equals(ip))
+ .findFirst()
+ .get();
+ } catch (SocketException e) {
+ throw new IllegalStateException("Cannot determine local IP", e);
+ }
+ }
+}
\ No newline at end of file