Fix for ONOS-5035.Cherry picked from commit f1621f163d1b48d7056ef676cdcece3822bde0f3
Change-Id: I99cce9565f479a03e9e472713efc6d4ddd749108
diff --git a/core/store/dist/src/main/java/org/onosproject/store/host/impl/DistributedHostStore.java b/core/store/dist/src/main/java/org/onosproject/store/host/impl/DistributedHostStore.java
index 7f9be04..bd99cc5 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/host/impl/DistributedHostStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/host/impl/DistributedHostStore.java
@@ -55,14 +55,19 @@
import org.onosproject.store.service.StorageService;
import org.onosproject.store.service.WallClockTimestamp;
import org.osgi.service.component.ComponentContext;
+import org.onosproject.store.service.DistributedPrimitive.Status;
import org.slf4j.Logger;
import java.util.Collection;
import java.util.Dictionary;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -70,6 +75,8 @@
import static org.onlab.util.Tools.get;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.groupedThreads;
import static org.onosproject.net.DefaultAnnotations.merge;
import static org.onosproject.net.host.HostEvent.Type.*;
import static org.slf4j.LoggerFactory.getLogger;
@@ -80,8 +87,8 @@
@Component(immediate = true)
@Service
public class DistributedHostStore
- extends AbstractStore<HostEvent, HostStoreDelegate>
- implements HostStore {
+ extends AbstractStore<HostEvent, HostStoreDelegate>
+ implements HostStore {
private final Logger log = getLogger(getClass());
@@ -99,12 +106,18 @@
private ConsistentMap<HostId, DefaultHost> hostsConsistentMap;
private Map<HostId, DefaultHost> hosts;
+ private Map<IpAddress, Set<Host>> hostsByIp;
private MapEventListener<HostId, DefaultHost> hostLocationTracker =
new HostLocationTracker();
+
private EventuallyConsistentMap<HostId, Timestamp> hostsTimestamp;
+ private ScheduledExecutorService executor;
+
+ private Consumer<Status> statusChangeListener;
+
@Activate
public void activate(ComponentContext context) {
configService.registerProperties(getClass());
@@ -126,7 +139,17 @@
.build();
hostsConsistentMap.addListener(hostLocationTracker);
+
modified(context);
+
+ executor = newSingleThreadScheduledExecutor(groupedThreads("onos/hosts", "store", log));
+ statusChangeListener = status -> {
+ if (status == Status.ACTIVE) {
+ executor.execute(this::loadHostsByIp);
+ }
+ };
+ hostsConsistentMap.addStatusChangeListener(statusChangeListener);
+ loadHostsByIp();
log.info("Started");
}
@@ -156,7 +179,19 @@
}
}
-
+ private void loadHostsByIp() {
+ hostsByIp = new ConcurrentHashMap<IpAddress, Set<Host>>();
+ hostsConsistentMap.asJavaMap().values().forEach(host -> {
+ host.ipAddresses().forEach(ip -> {
+ Set<Host> existingHosts = hostsByIp.get(ip);
+ if (existingHosts == null) {
+ hostsByIp.put(ip, addHosts(host));
+ } else {
+ existingHosts.add(host);
+ }
+ });
+ });
+ }
private boolean shouldUpdate(DefaultHost existingHost,
ProviderId providerId,
@@ -176,7 +211,7 @@
if (replaceIPs) {
if (!Objects.equals(hostDescription.ipAddress(),
- existingHost.ipAddresses())) {
+ existingHost.ipAddresses())) {
return true;
}
} else {
@@ -188,8 +223,8 @@
// check to see if any of the annotations provided by hostDescription
// differ from those in the existing host
return hostDescription.annotations().keys().stream()
- .anyMatch(k -> !Objects.equals(hostDescription.annotations().value(k),
- existingHost.annotations().value(k)));
+ .anyMatch(k -> !Objects.equals(hostDescription.annotations().value(k),
+ existingHost.annotations().value(k)));
}
@@ -205,39 +240,39 @@
}
hostsConsistentMap.computeIf(hostId,
- existingHost -> shouldUpdate(existingHost, providerId, hostId,
- hostDescription, replaceIPs),
- (id, existingHost) -> {
- HostLocation location = hostDescription.location();
+ existingHost -> shouldUpdate(existingHost, providerId, hostId,
+ hostDescription, replaceIPs),
+ (id, existingHost) -> {
+ HostLocation location = hostDescription.location();
- final Set<IpAddress> addresses;
- if (existingHost == null || replaceIPs) {
- addresses = ImmutableSet.copyOf(hostDescription.ipAddress());
- } else {
- addresses = Sets.newHashSet(existingHost.ipAddresses());
- addresses.addAll(hostDescription.ipAddress());
- }
+ final Set<IpAddress> addresses;
+ if (existingHost == null || replaceIPs) {
+ addresses = ImmutableSet.copyOf(hostDescription.ipAddress());
+ } else {
+ addresses = Sets.newHashSet(existingHost.ipAddresses());
+ addresses.addAll(hostDescription.ipAddress());
+ }
- final Annotations annotations;
- final boolean configured;
- if (existingHost != null) {
- annotations = merge((DefaultAnnotations) existingHost.annotations(),
- hostDescription.annotations());
- configured = existingHost.configured();
- } else {
- annotations = hostDescription.annotations();
- configured = hostDescription.configured();
- }
+ final Annotations annotations;
+ final boolean configured;
+ if (existingHost != null) {
+ annotations = merge((DefaultAnnotations) existingHost.annotations(),
+ hostDescription.annotations());
+ configured = existingHost.configured();
+ } else {
+ annotations = hostDescription.annotations();
+ configured = hostDescription.configured();
+ }
- return new DefaultHost(providerId,
- hostId,
- hostDescription.hwAddress(),
- hostDescription.vlan(),
- location,
- addresses,
- configured,
- annotations);
- });
+ return new DefaultHost(providerId,
+ hostId,
+ hostDescription.hwAddress(),
+ hostDescription.vlan(),
+ location,
+ addresses,
+ configured,
+ annotations);
+ });
return null;
}
@@ -262,6 +297,7 @@
if (addresses != null && addresses.contains(ipAddress)) {
addresses = new HashSet<>(existingHost.ipAddresses());
addresses.remove(ipAddress);
+ removeIpFromHostsByIp(existingHost, ipAddress);
return new DefaultHost(existingHost.providerId(),
hostId,
existingHost.mac(),
@@ -305,7 +341,8 @@
@Override
public Set<Host> getHosts(IpAddress ip) {
- return filter(hosts.values(), host -> host.ipAddresses().contains(ip));
+ Set<Host> hosts = hostsByIp.get(ip);
+ return hosts != null ? ImmutableSet.copyOf(hosts) : ImmutableSet.of();
}
@Override
@@ -335,18 +372,70 @@
return collection.stream().filter(predicate).collect(Collectors.toSet());
}
+ private Set<Host> addHosts(Host host) {
+ Set<Host> hosts = Sets.newConcurrentHashSet();
+ hosts.add(host);
+ return hosts;
+ }
+
+ private Set<Host> updateHosts(Set<Host> existingHosts, Host host) {
+ Iterator<Host> iterator = existingHosts.iterator();
+ while (iterator.hasNext()) {
+ Host existingHost = iterator.next();
+ if (existingHost.id().equals(host.id())) {
+ iterator.remove();
+ }
+ }
+ existingHosts.add(host);
+ return existingHosts;
+ }
+
+ private Set<Host> removeHosts(Set<Host> existingHosts, Host host) {
+ if (existingHosts != null) {
+ Iterator<Host> iterator = existingHosts.iterator();
+ while (iterator.hasNext()) {
+ Host existingHost = iterator.next();
+ if (existingHost.id().equals(host.id())) {
+ iterator.remove();
+ }
+ }
+ }
+
+ if (existingHosts.isEmpty()) {
+ return null;
+ }
+ return existingHosts;
+ }
+
+ private void updateHostsByIp(DefaultHost host) {
+ host.ipAddresses().forEach(ip -> {
+ hostsByIp.compute(ip, (k, v) -> v == null ? addHosts(host)
+ : updateHosts(v, host));
+ });
+ }
+
+ private void removeHostsByIp(DefaultHost host) {
+ host.ipAddresses().forEach(ip -> {
+ hostsByIp.computeIfPresent(ip, (k, v) -> removeHosts(v, host));
+ });
+ }
+
+ private void removeIpFromHostsByIp(DefaultHost host, IpAddress ip) {
+ hostsByIp.computeIfPresent(ip, (k, v) -> removeHosts(v, host));
+ }
+
private class HostLocationTracker implements MapEventListener<HostId, DefaultHost> {
@Override
public void event(MapEvent<HostId, DefaultHost> event) {
- Host host;
+ DefaultHost host = checkNotNull(event.value().value());
switch (event.type()) {
case INSERT:
- host = checkNotNull(event.newValue().value());
+ updateHostsByIp(host);
notifyDelegate(new HostEvent(HOST_ADDED, host));
break;
case UPDATE:
- host = checkNotNull(event.newValue().value());
- Host prevHost = checkNotNull(event.oldValue().value());
+ updateHostsByIp(host);
+ DefaultHost prevHost = checkNotNull(event.oldValue().value());
if (!Objects.equals(prevHost.location(), host.location())) {
notifyDelegate(new HostEvent(HOST_MOVED, host, prevHost));
} else if (!Objects.equals(prevHost, host)) {
@@ -354,7 +443,7 @@
}
break;
case REMOVE:
- host = checkNotNull(event.oldValue().value());
+ updateHostsByIp(host);
notifyDelegate(new HostEvent(HOST_REMOVED, host));
break;
default:
diff --git a/core/store/dist/src/test/java/org/onosproject/store/host/impl/DistributedHostStoreTest.java b/core/store/dist/src/test/java/org/onosproject/store/host/impl/DistributedHostStoreTest.java
index 9ca2c98..5157642 100644
--- a/core/store/dist/src/test/java/org/onosproject/store/host/impl/DistributedHostStoreTest.java
+++ b/core/store/dist/src/test/java/org/onosproject/store/host/impl/DistributedHostStoreTest.java
@@ -32,6 +32,8 @@
import org.onosproject.store.service.TestStorageService;
import java.util.Dictionary;
+import com.google.common.collect.Sets;
+
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Set;
@@ -47,6 +49,7 @@
private DistributedHostStore ecXHostStore;
private static final HostId HOSTID = HostId.hostId(MacAddress.valueOf("1a:1a:1a:1a:1a:1a"));
+ private static final HostId HOSTID1 = HostId.hostId(MacAddress.valueOf("1a:1a:1a:1a:1a:1b"));
private static final IpAddress IP1 = IpAddress.valueOf("10.2.0.2");
private static final IpAddress IP2 = IpAddress.valueOf("10.2.0.3");
@@ -85,10 +88,7 @@
ips.add(IP1);
ips.add(IP2);
- HostDescription description = new DefaultHostDescription(HOSTID.mac(),
- HOSTID.vlanId(),
- HostLocation.NONE,
- ips);
+ HostDescription description = createHostDesc(HOSTID, ips);
ecXHostStore.createOrUpdateHost(PID, HOSTID, description, false);
ecXHostStore.removeIp(HOSTID, IP1);
Host host = ecXHostStore.getHost(HOSTID);
@@ -104,9 +104,9 @@
ips.add(IP2);
HostDescription description = new DefaultHostDescription(HOSTID.mac(),
- HOSTID.vlanId(),
- HostLocation.NONE,
- ips);
+ HOSTID.vlanId(),
+ HostLocation.NONE,
+ ips);
ecXHostStore.createOrUpdateHost(PID, HOSTID, description, false);
Timestamp timestamp = ecXHostStore.getHostLastseenTime(HOSTID);
@@ -114,4 +114,49 @@
assertTrue(timestamp != null);
}
+ @Test
+ public void testAddHostByIp() {
+ Set<IpAddress> ips = new HashSet<>();
+ ips.add(IP1);
+ ips.add(IP2);
+
+ HostDescription description = createHostDesc(HOSTID, ips);
+ ecXHostStore.createOrUpdateHost(PID, HOSTID, description, false);
+
+ Set<Host> hosts = ecXHostStore.getHosts(IP1);
+
+ assertFalse(hosts.size() > 1);
+ assertTrue(hosts.size() == 1);
+
+ HostDescription description1 = createHostDesc(HOSTID1, Sets.newHashSet(IP2));
+ ecXHostStore.createOrUpdateHost(PID, HOSTID1, description1, false);
+
+ Set<Host> hosts1 = ecXHostStore.getHosts(IP2);
+
+ assertFalse(hosts1.size() < 1);
+ assertTrue(hosts1.size() == 2);
+ }
+
+ @Test
+ public void testRemoveHostByIp() {
+ Set<IpAddress> ips = new HashSet<>();
+ ips.add(IP1);
+ ips.add(IP2);
+
+ HostDescription description = createHostDesc(HOSTID, ips);
+ ecXHostStore.createOrUpdateHost(PID, HOSTID, description, false);
+ ecXHostStore.removeIp(HOSTID, IP1);
+ Set<Host> hosts = ecXHostStore.getHosts(IP1);
+ assertTrue(hosts.size() == 0);
+ }
+
+ private HostDescription createHostDesc(HostId hostId, Set<IpAddress> ips) {
+ return new DefaultHostDescription(hostId.mac(),
+ hostId.vlanId(),
+ HostLocation.NONE,
+ ips);
+ }
+
+
+
}