diff options
Diffstat (limited to 'framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl')
9 files changed, 0 insertions, 937 deletions
diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterManagementMessageSubjects.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterManagementMessageSubjects.java deleted file mode 100644 index 918f7921..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterManagementMessageSubjects.java +++ /dev/null @@ -1,26 +0,0 @@ -/* - * Copyright 2014 Open Networking Laboratory - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.onosproject.store.cluster.impl; - -import org.onosproject.store.cluster.messaging.MessageSubject; - -//Not used right now -public final class ClusterManagementMessageSubjects { - // avoid instantiation - private ClusterManagementMessageSubjects() {} - - public static final MessageSubject CLUSTER_MEMBERSHIP_EVENT = new MessageSubject("CLUSTER_MEMBERSHIP_EVENT"); -} diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterMembershipEvent.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterMembershipEvent.java deleted file mode 100644 index c6428739..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterMembershipEvent.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * Copyright 2014 Open Networking Laboratory - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.onosproject.store.cluster.impl; - -import org.onosproject.cluster.ControllerNode; - -//Not used right now -/** - * Contains information that will be published when a cluster membership event occurs. - */ -public class ClusterMembershipEvent { - - private final ClusterMembershipEventType type; - private final ControllerNode node; - - public ClusterMembershipEvent(ClusterMembershipEventType type, ControllerNode node) { - this.type = type; - this.node = node; - } - - public ClusterMembershipEventType type() { - return type; - } - - public ControllerNode node() { - return node; - } -} diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterMembershipEventType.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterMembershipEventType.java deleted file mode 100644 index a7f09c71..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterMembershipEventType.java +++ /dev/null @@ -1,24 +0,0 @@ -/* - * Copyright 2014 Open Networking Laboratory - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.onosproject.store.cluster.impl; - -//Not used right now -public enum ClusterMembershipEventType { - NEW_MEMBER, - LEAVING_MEMBER, - UNREACHABLE_MEMBER, - HEART_BEAT, -} diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterNodesDelegate.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterNodesDelegate.java deleted file mode 100644 index 7aeca72f..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterNodesDelegate.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Copyright 2014 Open Networking Laboratory - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.onosproject.store.cluster.impl; - -import org.onosproject.cluster.DefaultControllerNode; -import org.onosproject.cluster.NodeId; -import org.onlab.packet.IpAddress; - -// Not used right now -/** - * Simple back interface through which connection manager can interact with - * the cluster store. - */ -public interface ClusterNodesDelegate { - - /** - * Notifies about cluster node coming online. - * - * @param nodeId newly detected cluster node id - * @param ip node IP listen address - * @param tcpPort node TCP listen port - * @return the controller node - */ - DefaultControllerNode nodeDetected(NodeId nodeId, IpAddress ip, - int tcpPort); - - /** - * Notifies about cluster node going offline. - * - * @param nodeId identifier of the cluster node that vanished - */ - void nodeVanished(NodeId nodeId); - - /** - * Notifies about remote request to remove node from cluster. - * - * @param nodeId identifier of the cluster node that was removed - */ - void nodeRemoved(NodeId nodeId); - -} diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DistributedClusterStore.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DistributedClusterStore.java deleted file mode 100644 index 3bb6a708..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DistributedClusterStore.java +++ /dev/null @@ -1,277 +0,0 @@ -/* - * Copyright 2014-2015 Open Networking Laboratory - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.onosproject.store.cluster.impl; - -import com.google.common.collect.ImmutableSet; -import com.google.common.collect.Maps; - -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.joda.time.DateTime; -import org.onlab.packet.IpAddress; -import org.onlab.util.KryoNamespace; -import org.onosproject.cluster.ClusterEvent; -import org.onosproject.cluster.ClusterMetadataService; -import org.onosproject.cluster.ClusterStore; -import org.onosproject.cluster.ClusterStoreDelegate; -import org.onosproject.cluster.ControllerNode; -import org.onosproject.cluster.ControllerNode.State; -import org.onosproject.cluster.DefaultControllerNode; -import org.onosproject.cluster.NodeId; -import org.onosproject.store.AbstractStore; -import org.onosproject.store.cluster.messaging.Endpoint; -import org.onosproject.store.cluster.messaging.MessagingService; -import org.onosproject.store.serializers.KryoNamespaces; -import org.onosproject.store.serializers.KryoSerializer; -import org.slf4j.Logger; - -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.function.Consumer; -import java.util.stream.Collectors; - -import static com.google.common.base.Preconditions.checkNotNull; -import static org.onlab.util.Tools.groupedThreads; -import static org.slf4j.LoggerFactory.getLogger; - -@Component(immediate = true) -@Service -/** - * Distributed cluster nodes store that employs an accrual failure - * detector to identify cluster member up/down status. - */ -public class DistributedClusterStore - extends AbstractStore<ClusterEvent, ClusterStoreDelegate> - implements ClusterStore { - - private static final Logger log = getLogger(DistributedClusterStore.class); - - public static final String HEARTBEAT_MESSAGE = "onos-cluster-heartbeat"; - - // TODO: make these configurable. - private static final int HEARTBEAT_INTERVAL_MS = 100; - private static final int PHI_FAILURE_THRESHOLD = 10; - - private static final KryoSerializer SERIALIZER = new KryoSerializer() { - @Override - protected void setupKryoPool() { - serializerPool = KryoNamespace.newBuilder() - .register(KryoNamespaces.API) - .register(HeartbeatMessage.class) - .build() - .populate(1); - } - }; - - private static final String INSTANCE_ID_NULL = "Instance ID cannot be null"; - - private final Map<NodeId, ControllerNode> allNodes = Maps.newConcurrentMap(); - private final Map<NodeId, State> nodeStates = Maps.newConcurrentMap(); - private final Map<NodeId, DateTime> nodeStateLastUpdatedTimes = Maps.newConcurrentMap(); - private ScheduledExecutorService heartBeatSender = Executors.newSingleThreadScheduledExecutor( - groupedThreads("onos/cluster/membership", "heartbeat-sender")); - private ExecutorService heartBeatMessageHandler = Executors.newSingleThreadExecutor( - groupedThreads("onos/cluster/membership", "heartbeat-receiver")); - - private PhiAccrualFailureDetector failureDetector; - - private ControllerNode localNode; - - @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) - protected ClusterMetadataService clusterMetadataService; - - @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) - protected MessagingService messagingService; - - @Activate - public void activate() { - localNode = clusterMetadataService.getLocalNode(); - - messagingService.registerHandler(HEARTBEAT_MESSAGE, - new HeartbeatMessageHandler(), heartBeatMessageHandler); - - failureDetector = new PhiAccrualFailureDetector(); - - heartBeatSender.scheduleWithFixedDelay(this::heartbeat, 0, - HEARTBEAT_INTERVAL_MS, TimeUnit.MILLISECONDS); - - log.info("Started"); - } - - @Deactivate - public void deactivate() { - messagingService.unregisterHandler(HEARTBEAT_MESSAGE); - heartBeatSender.shutdownNow(); - heartBeatMessageHandler.shutdownNow(); - - log.info("Stopped"); - } - - @Override - public void setDelegate(ClusterStoreDelegate delegate) { - checkNotNull(delegate, "Delegate cannot be null"); - this.delegate = delegate; - } - - @Override - public void unsetDelegate(ClusterStoreDelegate delegate) { - this.delegate = null; - } - - @Override - public boolean hasDelegate() { - return this.delegate != null; - } - - @Override - public ControllerNode getLocalNode() { - return localNode; - } - - @Override - public Set<ControllerNode> getNodes() { - return ImmutableSet.copyOf(allNodes.values()); - } - - @Override - public ControllerNode getNode(NodeId nodeId) { - checkNotNull(nodeId, INSTANCE_ID_NULL); - return allNodes.get(nodeId); - } - - @Override - public State getState(NodeId nodeId) { - checkNotNull(nodeId, INSTANCE_ID_NULL); - return nodeStates.get(nodeId); - } - - @Override - public ControllerNode addNode(NodeId nodeId, IpAddress ip, int tcpPort) { - ControllerNode node = new DefaultControllerNode(nodeId, ip, tcpPort); - addNode(node); - return node; - } - - @Override - public void removeNode(NodeId nodeId) { - checkNotNull(nodeId, INSTANCE_ID_NULL); - ControllerNode node = allNodes.remove(nodeId); - if (node != null) { - nodeStates.remove(nodeId); - notifyDelegate(new ClusterEvent(ClusterEvent.Type.INSTANCE_REMOVED, node)); - } - } - - private void addNode(ControllerNode node) { - allNodes.put(node.id(), node); - updateState(node.id(), node.equals(localNode) ? State.ACTIVE : State.INACTIVE); - notifyDelegate(new ClusterEvent(ClusterEvent.Type.INSTANCE_ADDED, node)); - } - - private void updateState(NodeId nodeId, State newState) { - nodeStates.put(nodeId, newState); - nodeStateLastUpdatedTimes.put(nodeId, DateTime.now()); - } - - private void heartbeat() { - try { - Set<ControllerNode> peers = allNodes.values() - .stream() - .filter(node -> !(node.id().equals(localNode.id()))) - .collect(Collectors.toSet()); - byte[] hbMessagePayload = SERIALIZER.encode(new HeartbeatMessage(localNode, peers)); - peers.forEach((node) -> { - heartbeatToPeer(hbMessagePayload, node); - State currentState = nodeStates.get(node.id()); - double phi = failureDetector.phi(node.id()); - if (phi >= PHI_FAILURE_THRESHOLD) { - if (currentState == State.ACTIVE) { - updateState(node.id(), State.INACTIVE); - notifyStateChange(node.id(), State.ACTIVE, State.INACTIVE); - } - } else { - if (currentState == State.INACTIVE) { - updateState(node.id(), State.ACTIVE); - notifyStateChange(node.id(), State.INACTIVE, State.ACTIVE); - } - } - }); - } catch (Exception e) { - log.debug("Failed to send heartbeat", e); - } - } - - private void notifyStateChange(NodeId nodeId, State oldState, State newState) { - ControllerNode node = allNodes.get(nodeId); - if (newState == State.ACTIVE) { - notifyDelegate(new ClusterEvent(ClusterEvent.Type.INSTANCE_ACTIVATED, node)); - } else { - notifyDelegate(new ClusterEvent(ClusterEvent.Type.INSTANCE_DEACTIVATED, node)); - } - } - - private void heartbeatToPeer(byte[] messagePayload, ControllerNode peer) { - Endpoint remoteEp = new Endpoint(peer.ip(), peer.tcpPort()); - messagingService.sendAsync(remoteEp, HEARTBEAT_MESSAGE, messagePayload).whenComplete((result, error) -> { - if (error != null) { - log.trace("Sending heartbeat to {} failed", remoteEp, error); - } - }); - } - - private class HeartbeatMessageHandler implements Consumer<byte[]> { - @Override - public void accept(byte[] message) { - HeartbeatMessage hb = SERIALIZER.decode(message); - failureDetector.report(hb.source().id()); - hb.knownPeers().forEach(node -> { - allNodes.put(node.id(), node); - }); - } - } - - private static class HeartbeatMessage { - private ControllerNode source; - private Set<ControllerNode> knownPeers; - - public HeartbeatMessage(ControllerNode source, Set<ControllerNode> members) { - this.source = source; - this.knownPeers = ImmutableSet.copyOf(members); - } - - public ControllerNode source() { - return source; - } - - public Set<ControllerNode> knownPeers() { - return knownPeers; - } - } - - @Override - public DateTime getLastUpdated(NodeId nodeId) { - return nodeStateLastUpdatedTimes.get(nodeId); - } - -} diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/NodeInfo.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/NodeInfo.java deleted file mode 100644 index d436ca76..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/NodeInfo.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * Copyright 2015 Open Networking Laboratory - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.onosproject.store.cluster.impl; - -import static com.google.common.base.MoreObjects.toStringHelper; - -import java.util.Objects; - -import org.onosproject.cluster.ControllerNode; - -/** - * Node info read from configuration files during bootstrap. - */ -public final class NodeInfo { - private final String id; - private final String ip; - private final int tcpPort; - - private NodeInfo(String id, String ip, int port) { - this.id = id; - this.ip = ip; - this.tcpPort = port; - } - - /* - * Needed for serialization. - */ - private NodeInfo() { - id = null; - ip = null; - tcpPort = 0; - } - - /** - * Creates a new instance. - * @param id node id - * @param ip node ip address - * @param port tcp port - * @return NodeInfo - */ - public static NodeInfo from(String id, String ip, int port) { - NodeInfo node = new NodeInfo(id, ip, port); - return node; - } - - /** - * Returns the NodeInfo for a controller node. - * @param node controller node - * @return NodeInfo - */ - public static NodeInfo of(ControllerNode node) { - return NodeInfo.from(node.id().toString(), node.ip().toString(), node.tcpPort()); - } - - /** - * Returns node id. - * @return node id - */ - public String getId() { - return id; - } - - /** - * Returns node ip. - * @return node ip - */ - public String getIp() { - return ip; - } - - /** - * Returns node port. - * @return port - */ - public int getTcpPort() { - return tcpPort; - } - - @Override - public int hashCode() { - return Objects.hash(id, ip, tcpPort); - } - - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o instanceof NodeInfo) { - NodeInfo that = (NodeInfo) o; - return Objects.equals(this.id, that.id) && - Objects.equals(this.ip, that.ip) && - Objects.equals(this.tcpPort, that.tcpPort); - } - return false; - } - - @Override - public String toString() { - return toStringHelper(this) - .add("id", id) - .add("ip", ip) - .add("tcpPort", tcpPort).toString(); - } -}
\ No newline at end of file diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/PhiAccrualFailureDetector.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/PhiAccrualFailureDetector.java deleted file mode 100644 index cdb138b4..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/PhiAccrualFailureDetector.java +++ /dev/null @@ -1,119 +0,0 @@ -/* - * Copyright 2015 Open Networking Laboratory - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.onosproject.store.cluster.impl; - -import static com.google.common.base.Preconditions.checkArgument; -import static com.google.common.base.Preconditions.checkNotNull; - -import java.util.Map; - -import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics; -import org.onosproject.cluster.NodeId; - -import com.google.common.collect.Maps; - -/** - * Phi Accrual failure detector. - * <p> - * Based on a paper titled: "The φ Accrual Failure Detector" by Hayashibara, et al. - */ -public class PhiAccrualFailureDetector { - private final Map<NodeId, History> states = Maps.newConcurrentMap(); - - // TODO: make these configurable. - private static final int WINDOW_SIZE = 250; - private static final int MIN_SAMPLES = 25; - private static final double PHI_FACTOR = 1.0 / Math.log(10.0); - - // If a node does not have any heartbeats, this is the phi - // value to report. Indicates the node is inactive (from the - // detectors perspective. - private static final double BOOTSTRAP_PHI_VALUE = 100.0; - - /** - * Report a new heart beat for the specified node id. - * @param nodeId node id - */ - public void report(NodeId nodeId) { - report(nodeId, System.currentTimeMillis()); - } - - /** - * Report a new heart beat for the specified node id. - * @param nodeId node id - * @param arrivalTime arrival time - */ - public void report(NodeId nodeId, long arrivalTime) { - checkNotNull(nodeId, "NodeId must not be null"); - checkArgument(arrivalTime >= 0, "arrivalTime must not be negative"); - History nodeState = - states.computeIfAbsent(nodeId, key -> new History()); - synchronized (nodeState) { - long latestHeartbeat = nodeState.latestHeartbeatTime(); - if (latestHeartbeat != -1) { - nodeState.samples().addValue(arrivalTime - latestHeartbeat); - } - nodeState.setLatestHeartbeatTime(arrivalTime); - } - } - - /** - * Compute phi for the specified node id. - * @param nodeId node id - * @return phi value - */ - public double phi(NodeId nodeId) { - checkNotNull(nodeId, "NodeId must not be null"); - if (!states.containsKey(nodeId)) { - return BOOTSTRAP_PHI_VALUE; - } - History nodeState = states.get(nodeId); - synchronized (nodeState) { - long latestHeartbeat = nodeState.latestHeartbeatTime(); - DescriptiveStatistics samples = nodeState.samples(); - if (latestHeartbeat == -1 || samples.getN() < MIN_SAMPLES) { - return 0.0; - } - return computePhi(samples, latestHeartbeat, System.currentTimeMillis()); - } - } - - private double computePhi(DescriptiveStatistics samples, long tLast, long tNow) { - long size = samples.getN(); - long t = tNow - tLast; - return (size > 0) - ? PHI_FACTOR * t / samples.getMean() - : BOOTSTRAP_PHI_VALUE; - } - - private static class History { - DescriptiveStatistics samples = - new DescriptiveStatistics(WINDOW_SIZE); - long lastHeartbeatTime = -1; - - public DescriptiveStatistics samples() { - return samples; - } - - public long latestHeartbeatTime() { - return lastHeartbeatTime; - } - - public void setLatestHeartbeatTime(long value) { - lastHeartbeatTime = value; - } - } -}
\ No newline at end of file diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/StaticClusterMetadataStore.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/StaticClusterMetadataStore.java deleted file mode 100644 index 3cd992bb..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/StaticClusterMetadataStore.java +++ /dev/null @@ -1,258 +0,0 @@ -package org.onosproject.store.cluster.impl; - -import static com.google.common.base.Preconditions.checkNotNull; -import static java.net.NetworkInterface.getNetworkInterfaces; -import static org.slf4j.LoggerFactory.getLogger; - -import java.io.File; -import java.io.IOException; -import java.net.InetAddress; -import java.net.Inet4Address; -import java.net.NetworkInterface; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Function; - -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.Service; -import org.onlab.packet.IpAddress; -import org.onosproject.cluster.ClusterMetadata; -import org.onosproject.cluster.ClusterMetadataEvent; -import org.onosproject.cluster.ClusterMetadataStore; -import org.onosproject.cluster.ClusterMetadataStoreDelegate; -import org.onosproject.cluster.ControllerNode; -import org.onosproject.cluster.DefaultControllerNode; -import org.onosproject.cluster.NodeId; -import org.onosproject.cluster.Partition; -import org.onosproject.store.AbstractStore; -import org.onosproject.store.service.Versioned; -import org.slf4j.Logger; - -import com.fasterxml.jackson.core.JsonGenerator; -import com.fasterxml.jackson.core.JsonParser; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.DeserializationContext; -import com.fasterxml.jackson.databind.JsonDeserializer; -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.JsonSerializer; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.SerializerProvider; -import com.fasterxml.jackson.databind.module.SimpleModule; -import com.google.common.base.Throwables; -import com.google.common.collect.Lists; -import com.google.common.io.Files; - -/** - * ClusterMetadataStore backed by a local file. - */ -@Component(immediate = true, enabled = true) -@Service -public class StaticClusterMetadataStore - extends AbstractStore<ClusterMetadataEvent, ClusterMetadataStoreDelegate> - implements ClusterMetadataStore { - - private final Logger log = getLogger(getClass()); - - private static final String ONOS_IP = "ONOS_IP"; - private static final String ONOS_INTERFACE = "ONOS_INTERFACE"; - private static final String ONOS_ALLOW_IPV6 = "ONOS_ALLOW_IPV6"; - private static final String DEFAULT_ONOS_INTERFACE = "eth0"; - private static final String CLUSTER_METADATA_FILE = "../config/cluster.json"; - private static final int DEFAULT_ONOS_PORT = 9876; - private final File metadataFile = new File(CLUSTER_METADATA_FILE); - private AtomicReference<ClusterMetadata> metadata = new AtomicReference<>(); - private ObjectMapper mapper; - private long version; - - @Activate - public void activate() { - mapper = new ObjectMapper(); - SimpleModule module = new SimpleModule(); - module.addSerializer(NodeId.class, new NodeIdSerializer()); - module.addDeserializer(NodeId.class, new NodeIdDeserializer()); - module.addSerializer(ControllerNode.class, new ControllerNodeSerializer()); - module.addDeserializer(ControllerNode.class, new ControllerNodeDeserializer()); - mapper.registerModule(module); - File metadataFile = new File(CLUSTER_METADATA_FILE); - if (metadataFile.exists()) { - try { - metadata.set(mapper.readValue(metadataFile, ClusterMetadata.class)); - version = metadataFile.lastModified(); - } catch (IOException e) { - Throwables.propagate(e); - } - } else { - String localIp = getSiteLocalAddress(); - ControllerNode localNode = - new DefaultControllerNode(new NodeId(localIp), IpAddress.valueOf(localIp), DEFAULT_ONOS_PORT); - metadata.set(ClusterMetadata.builder() - .withName("default") - .withControllerNodes(Arrays.asList(localNode)) - .withPartitions(Lists.newArrayList(new Partition("p1", Lists.newArrayList(localNode.id())))) - .build()); - version = System.currentTimeMillis(); - } - log.info("Started"); - } - - @Deactivate - public void deactivate() { - log.info("Stopped"); - } - - @Override - public void setDelegate(ClusterMetadataStoreDelegate delegate) { - checkNotNull(delegate, "Delegate cannot be null"); - this.delegate = delegate; - } - - @Override - public void unsetDelegate(ClusterMetadataStoreDelegate delegate) { - this.delegate = null; - } - - @Override - public boolean hasDelegate() { - return this.delegate != null; - } - - @Override - public Versioned<ClusterMetadata> getClusterMetadata() { - return new Versioned<>(metadata.get(), version); - } - - @Override - public void setClusterMetadata(ClusterMetadata metadata) { - checkNotNull(metadata); - try { - Files.createParentDirs(metadataFile); - mapper.writeValue(metadataFile, metadata); - this.metadata.set(metadata); - } catch (IOException e) { - Throwables.propagate(e); - } - } - - @Override - public void setActiveReplica(String partitionId, NodeId nodeId) { - throw new UnsupportedOperationException(); - } - - @Override - public void unsetActiveReplica(String partitionId, NodeId nodeId) { - throw new UnsupportedOperationException(); - } - - @Override - public Collection<NodeId> getActiveReplicas(String partitionId) { - return metadata.get().getPartitions() - .stream() - .filter(r -> r.getName().equals(partitionId)) - .findFirst() - .map(r -> r.getMembers()) - .orElse(null); - } - - private static class ControllerNodeSerializer extends JsonSerializer<ControllerNode> { - @Override - public void serialize(ControllerNode node, JsonGenerator jgen, SerializerProvider provider) - throws IOException, JsonProcessingException { - jgen.writeStartObject(); - jgen.writeStringField("id", node.id().toString()); - jgen.writeStringField("ip", node.ip().toString()); - jgen.writeNumberField("port", node.tcpPort()); - jgen.writeEndObject(); - } - } - - private static class ControllerNodeDeserializer extends JsonDeserializer<ControllerNode> { - @Override - public ControllerNode deserialize(JsonParser jp, DeserializationContext ctxt) - throws IOException, JsonProcessingException { - JsonNode node = jp.getCodec().readTree(jp); - NodeId nodeId = new NodeId(node.get("id").textValue()); - IpAddress ip = IpAddress.valueOf(node.get("ip").textValue()); - int port = node.get("port").asInt(); - return new DefaultControllerNode(nodeId, ip, port); - } - } - - private static class NodeIdSerializer extends JsonSerializer<NodeId> { - @Override - public void serialize(NodeId nodeId, JsonGenerator jgen, SerializerProvider provider) - throws IOException, JsonProcessingException { - jgen.writeString(nodeId.toString()); - } - } - - private class NodeIdDeserializer extends JsonDeserializer<NodeId> { - @Override - public NodeId deserialize(JsonParser jp, DeserializationContext ctxt) - throws IOException, JsonProcessingException { - JsonNode node = jp.getCodec().readTree(jp); - return new NodeId(node.asText()); - } - } - - - private static String getSiteLocalAddress() { - - /* - * If the IP ONOS should use is set via the environment variable we will assume it is valid and should be used. - * Setting the IP address takes presidence over setting the interface via the environment. - */ - String useOnosIp = System.getenv(ONOS_IP); - if (useOnosIp != null) { - return useOnosIp; - } - - // Read environment variables for IP interface information or set to default - String useOnosInterface = System.getenv(ONOS_INTERFACE); - if (useOnosInterface == null) { - useOnosInterface = DEFAULT_ONOS_INTERFACE; - } - - // Capture if they want to limit IP address selection to only IPv4 (default). - boolean allowIPv6 = (System.getenv(ONOS_ALLOW_IPV6) != null); - - Function<NetworkInterface, IpAddress> ipLookup = nif -> { - IpAddress fallback = null; - - // nif can be null if the interface name specified doesn't exist on the node's host - if (nif != null) { - for (InetAddress address : Collections.list(nif.getInetAddresses())) { - if (address.isSiteLocalAddress() && (allowIPv6 || address instanceof Inet4Address)) { - return IpAddress.valueOf(address); - } - if (fallback == null && !address.isLoopbackAddress() && !address.isMulticastAddress() - && (allowIPv6 || address instanceof Inet4Address)) { - fallback = IpAddress.valueOf(address); - } - } - } - return fallback; - }; - try { - IpAddress ip = ipLookup.apply(NetworkInterface.getByName(useOnosInterface)); - if (ip != null) { - return ip.toString(); - } - for (NetworkInterface nif : Collections.list(getNetworkInterfaces())) { - if (!nif.getName().equals(useOnosInterface)) { - ip = ipLookup.apply(nif); - if (ip != null) { - return ip.toString(); - } - } - } - } catch (Exception e) { - throw new IllegalStateException("Unable to get network interfaces", e); - } - - return IpAddress.valueOf(InetAddress.getLoopbackAddress()).toString(); - } -} diff --git a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/package-info.java b/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/package-info.java deleted file mode 100644 index 9e2db676..00000000 --- a/framework/src/onos/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/package-info.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * Copyright 2014 Open Networking Laboratory - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/** - * Implementation of a distributed cluster node store using Hazelcast. - */ -package org.onosproject.store.cluster.impl; |