Implement host probing retry with major refactoring
- Implement probe retry
- Switch to typical core/provider design pattern for HostProbingService
and as a result decoupling the dependency between SR and HostLocationProvider
Change-Id: I33a15af580677ea376b421ac3e26f9821dcca844
diff --git a/core/store/dist/src/main/java/org/onosproject/store/host/impl/DefaultHostProbeStore.java b/core/store/dist/src/main/java/org/onosproject/store/host/impl/DefaultHostProbeStore.java
new file mode 100644
index 0000000..33cb7ad
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/host/impl/DefaultHostProbeStore.java
@@ -0,0 +1,199 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+ *
+ * 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.host.impl;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalNotification;
+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.MacAddress;
+import org.onlab.util.KryoNamespace;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.Host;
+import org.onosproject.net.host.HostProbe;
+import org.onosproject.net.host.HostProbeStore;
+import org.onosproject.net.host.HostProbingEvent;
+import org.onosproject.net.host.ProbeMode;
+import org.onosproject.net.host.HostProbingStoreDelegate;
+import org.onosproject.store.AbstractStore;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.AtomicCounter;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Versioned;
+import org.slf4j.Logger;
+
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.slf4j.LoggerFactory.getLogger;
+
+@Component(immediate = true)
+@Service
+public class DefaultHostProbeStore extends AbstractStore<HostProbingEvent, HostProbingStoreDelegate>
+ implements HostProbeStore {
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected StorageService storageService;
+
+ private final Logger log = getLogger(getClass());
+
+ // TODO make this configurable
+ private static final int PROBE_TIMEOUT_MS = 3000;
+
+ private AtomicCounter hostProbeIndex;
+ private Cache<MacAddress, HostProbe> probingHostsCache;
+ private ConsistentMap<MacAddress, HostProbe> probingHostsConsistentMap;
+ private Map<MacAddress, HostProbe> probingHosts;
+ private MapEventListener<MacAddress, HostProbe> probingHostListener = new ProbingHostListener();
+ private ScheduledExecutorService cacheCleaner;
+ private ScheduledExecutorService locationRemover;
+
+ @Activate
+ public void activate() {
+ KryoNamespace.Builder pendingHostSerializer = KryoNamespace.newBuilder()
+ .register(KryoNamespaces.API)
+ .register(DefaultHostProbe.class)
+ .register(ProbeMode.class);
+ probingHostsConsistentMap = storageService.<MacAddress, HostProbe>consistentMapBuilder()
+ .withName("onos-hosts-pending")
+ .withRelaxedReadConsistency()
+ .withSerializer(Serializer.using(pendingHostSerializer.build()))
+ .build();
+ probingHostsConsistentMap.addListener(probingHostListener);
+ probingHosts = probingHostsConsistentMap.asJavaMap();
+
+ hostProbeIndex = storageService.atomicCounterBuilder()
+ .withName("onos-hosts-probe-index")
+ .build()
+ .asAtomicCounter();
+
+ probingHostsCache = CacheBuilder.newBuilder()
+ .expireAfterWrite(PROBE_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+ .removalListener((RemovalNotification<MacAddress, HostProbe> notification) -> {
+ MacAddress probeMac = notification.getKey();
+ switch (notification.getCause()) {
+ case EXPIRED:
+ case REPLACED:
+ probingHosts.computeIfPresent(probeMac, (k, v) -> {
+ v.decreaseRetry();
+ return v;
+ });
+ break;
+ case EXPLICIT:
+ break;
+ default:
+ log.warn("Remove {} from pendingHostLocations for unexpected reason {}",
+ notification.getKey(), notification.getCause());
+ }
+ }).build();
+
+ cacheCleaner = newSingleThreadScheduledExecutor(
+ groupedThreads("onos/host/hostprobestore", "cache-cleaner", log));
+ cacheCleaner.scheduleAtFixedRate(probingHostsCache::cleanUp, 0, PROBE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ locationRemover = newSingleThreadScheduledExecutor(
+ groupedThreads("onos/host/hostprobestore", "loc-remover", log));
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ cacheCleaner.shutdown();
+ locationRemover.shutdown();
+ probingHostsCache.cleanUp();
+
+ log.info("Stopped");
+ }
+
+ @Override
+ public MacAddress addProbingHost(Host host, ConnectPoint connectPoint, ProbeMode probeMode,
+ MacAddress probeMac, int retry) {
+ if (probeMac == null) {
+ probeMac = generateProbeMac();
+ }
+ DefaultHostProbe probingHost = new DefaultHostProbe(host, connectPoint, probeMode, probeMac, retry);
+ probingHostsCache.put(probeMac, probingHost);
+ probingHosts.put(probeMac, probingHost);
+ return probeMac;
+ }
+
+ @Override
+ public void removeProbingHost(MacAddress probeMac) {
+ probingHostsCache.invalidate(probeMac);
+ probingHosts.remove(probeMac);
+ }
+
+ private MacAddress generateProbeMac() {
+ // Use ONLab OUI (3 bytes) + atomic counter (3 bytes) as the source MAC of the probe
+ long nextIndex = hostProbeIndex.incrementAndGet();
+ return MacAddress.valueOf(MacAddress.NONE.toLong() + nextIndex);
+ }
+
+ private class ProbingHostListener implements MapEventListener<MacAddress, HostProbe> {
+ @Override
+ public void event(MapEvent<MacAddress, HostProbe> event) {
+ HostProbe newValue = Versioned.valueOrNull(event.newValue());
+ HostProbe oldValue = Versioned.valueOrNull(event.oldValue());
+
+ HostProbingEvent hostProbingEvent;
+ switch (event.type()) {
+ case INSERT:
+ hostProbingEvent = new HostProbingEvent(HostProbingEvent.Type.PROBE_REQUESTED, newValue);
+ notifyDelegate(hostProbingEvent);
+ break;
+ case UPDATE:
+ // Fail VERIFY probe immediately. Only allow DISCOVER probe to retry.
+ if (newValue.retry() > 0) {
+ if (newValue.mode() == ProbeMode.DISCOVER) {
+ hostProbingEvent = new HostProbingEvent(HostProbingEvent.Type.PROBE_TIMEOUT,
+ newValue, oldValue);
+ notifyDelegate(hostProbingEvent);
+ } else {
+ hostProbingEvent = new HostProbingEvent(HostProbingEvent.Type.PROBE_FAIL,
+ newValue, oldValue);
+ notifyDelegate(hostProbingEvent);
+ }
+ } else {
+ // Remove from pendingHost and let the remove listener generates the event
+ locationRemover.execute(() -> probingHosts.remove(event.key()));
+ }
+ break;
+ case REMOVE:
+ if (oldValue.retry() > 0) {
+ hostProbingEvent = new HostProbingEvent(HostProbingEvent.Type.PROBE_COMPLETED, oldValue);
+ notifyDelegate(hostProbingEvent);
+ } else {
+ hostProbingEvent = new HostProbingEvent(HostProbingEvent.Type.PROBE_FAIL, oldValue);
+ notifyDelegate(hostProbingEvent);
+ }
+ break;
+ default:
+ log.warn("Unknown map event type: {}", event.type());
+ }
+ }
+ }
+}