Fix for ONOS-5034

Change-Id: I61158511680851be375b93c365fc32c0573e9edc

@@ -8,6 +8,7 @@ TEST_DEPS = [
     '//core/common:onos-core-common',
     '//core/store/dist:onos-core-dist',
     '//core/store/dist:onos-core-dist-tests',
+    '//utils/osgi:onlab-osgi-tests',
 ]
 
 osgi_jar_with_tests (

@@ -110,6 +110,13 @@
             <groupId>org.apache.felix</groupId>
             <artifactId>org.apache.felix.scr</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onlab-osgi</artifactId>
+            <version>${project.version}</version>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

@@ -18,12 +18,16 @@ package org.onosproject.net.host.impl;
 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.Modified;
+import org.apache.felix.scr.annotations.Property;
 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.onlab.packet.MacAddress;
 import org.onlab.packet.VlanId;
+import org.onlab.util.Tools;
+import org.onosproject.cfg.ComponentConfigService;
 import org.onosproject.incubator.net.intf.InterfaceService;
 import org.onosproject.net.edge.EdgePortService;
 import org.onosproject.net.provider.AbstractListenerProviderRegistry;
@@ -48,8 +52,10 @@ import org.onosproject.net.host.HostStore;
 import org.onosproject.net.host.HostStoreDelegate;
 import org.onosproject.net.packet.PacketService;
 import org.onosproject.net.provider.AbstractProviderService;
+import org.osgi.service.component.ComponentContext;
 import org.slf4j.Logger;
 
+import java.util.Dictionary;
 import java.util.Set;
 
 import static com.google.common.base.Preconditions.checkNotNull;
@@ -93,15 +99,29 @@ public class HostManager
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected EdgePortService edgePortService;
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ComponentConfigService cfgService;
+
+    @Property(name = "monitorHosts", boolValue = false,
+            label = "Enable/Disable monitoring of hosts")
+    private boolean monitorHosts = false;
+
+    @Property(name = "probeRate", longValue = 30000,
+            label = "Set the probe Rate in milli seconds")
+    private long probeRate = 30000;
+
     private HostMonitor monitor;
 
     @Activate
-    public void activate() {
+    public void activate(ComponentContext context) {
         store.setDelegate(delegate);
         eventDispatcher.addSink(HostEvent.class, listenerRegistry);
         networkConfigService.addListener(networkConfigListener);
         monitor = new HostMonitor(packetService, this, interfaceService, edgePortService);
+        monitor.setProbeRate(probeRate);
         monitor.start();
+        modified(context);
+        cfgService.registerProperties(getClass());
         log.info("Started");
     }
 
@@ -110,9 +130,82 @@ public class HostManager
         store.unsetDelegate(delegate);
         eventDispatcher.removeSink(HostEvent.class);
         networkConfigService.removeListener(networkConfigListener);
+        cfgService.unregisterProperties(getClass(), false);
+        monitor.shutdown();
         log.info("Stopped");
     }
 
+    @Modified
+    public void modified(ComponentContext context) {
+        boolean oldValue = monitorHosts;
+        readComponentConfiguration(context);
+        if (probeRate > 0) {
+            monitor.setProbeRate(probeRate);
+        } else {
+            log.warn("probeRate cannot be lessthan 0");
+        }
+        if (oldValue != monitorHosts) {
+            if (monitorHosts) {
+                startMonitoring();
+            } else {
+                stopMonitoring();
+            }
+        }
+    }
+
+    /**
+     * Extracts properties from the component configuration context.
+     *
+     * @param context the component context
+     */
+    private void readComponentConfiguration(ComponentContext context) {
+        Dictionary<?, ?> properties = context.getProperties();
+        Boolean flag;
+
+        flag = Tools.isPropertyEnabled(properties, "monitorHosts");
+        if (flag == null) {
+            log.info("monitorHosts is not enabled " +
+                             "using current value of {}", monitorHosts);
+        } else {
+            monitorHosts = flag;
+            log.info("Configured. monitorHosts {}",
+            monitorHosts ? "enabled" : "disabled");
+        }
+
+        Long longValue = Tools.getLongProperty(properties, "probeRate");
+        if (longValue == null || longValue == 0) {
+            log.info("probeRate is not set sing default value of {}", probeRate);
+        } else {
+            probeRate = longValue;
+            log.info("Configured. probeRate {}", probeRate);
+        }
+
+    }
+
+    /**
+     * Starts monitoring the hosts by IP Address.
+     *
+     */
+    private void startMonitoring() {
+        store.getHosts().forEach(host -> {
+                    host.ipAddresses().forEach(ip -> {
+                           monitor.addMonitoringFor(ip);
+            });
+        });
+    }
+
+    /**
+     * Stops monitoring the hosts by IP Address.
+     *
+     */
+    private void stopMonitoring() {
+        store.getHosts().forEach(host -> {
+                    host.ipAddresses().forEach(ip -> {
+                           monitor.stopMonitoring(ip);
+            });
+        });
+    }
+
     @Override
     protected HostProviderService createProviderService(HostProvider provider) {
         monitor.registerHostProvider(provider);
@@ -211,6 +304,11 @@ public class HostManager
             hostDescription = validateHost(hostDescription, hostId);
             store.createOrUpdateHost(provider().id(), hostId,
                                                        hostDescription, replaceIps);
+            if (monitorHosts) {
+                hostDescription.ipAddress().forEach(ip -> {
+                    monitor.addMonitoringFor(ip);
+                });
+            }
         }
 
         // returns a HostDescription made from the union of the BasicHostConfig
@@ -226,6 +324,12 @@ public class HostManager
         public void hostVanished(HostId hostId) {
             checkNotNull(hostId, HOST_ID_NULL);
             checkValidity();
+            Host host = store.getHost(hostId);
+            if (monitorHosts) {
+                host.ipAddresses().forEach(ip -> {
+                    monitor.stopMonitoring(ip);
+                });
+            }
             store.removeHost(hostId);
         }
 

@@ -139,6 +139,13 @@ public class HostMonitor implements TimerTask {
         }
     }
 
+    /*
+     * Sets the probe rate.
+     */
+    void setProbeRate(long probeRate) {
+        this.probeRate = probeRate;
+    }
+
     /**
      * Registers a host provider with the host monitor. The monitor can use the
      * provider to probe hosts.

@@ -21,9 +21,11 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.onlab.junit.TestTools;
+import org.onlab.osgi.ComponentContextAdapter;
 import org.onlab.packet.IpAddress;
 import org.onlab.packet.MacAddress;
 import org.onlab.packet.VlanId;
+import org.onosproject.cfg.ComponentConfigAdapter;
 import org.onosproject.common.event.impl.TestEventDispatcher;
 import org.onosproject.event.Event;
 import org.onosproject.net.DeviceId;
@@ -43,6 +45,8 @@ import org.onosproject.net.provider.AbstractProvider;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.store.trivial.SimpleHostStore;
 
+import java.util.Dictionary;
+import java.util.Hashtable;
 import java.util.List;
 import java.util.Set;
 
@@ -87,6 +91,16 @@ public class HostManagerTest {
     private static final HostLocation LOC1 = new HostLocation(DID1, P1, 123L);
     private static final HostLocation LOC2 = new HostLocation(DID1, P2, 123L);
 
+    public static final ComponentContextAdapter CTX_FOR_MONITOR = new ComponentContextAdapter() {
+        @Override
+        public Dictionary getProperties() {
+            Hashtable<String, String> props = new Hashtable<String, String>();
+            props.put("monitorHosts", "true");
+            props.put("probeRate", "40000");
+            return props;
+        }
+    };
+
     private HostManager mgr;
 
     protected TestListener listener = new TestListener();
@@ -101,7 +115,8 @@ public class HostManagerTest {
         injectEventDispatcher(mgr, new TestEventDispatcher());
         registry = mgr;
         mgr.networkConfigService = new TestNetworkConfigService();
-        mgr.activate();
+        mgr.cfgService = new ComponentConfigAdapter();
+        mgr.activate(CTX_FOR_MONITOR);
 
         mgr.addListener(listener);
 

@@ -44,17 +44,22 @@ import org.onosproject.net.Device;
 import org.onosproject.net.Host;
 import org.onosproject.net.HostId;
 import org.onosproject.net.HostLocation;
+import org.onosproject.net.MastershipRole;
 import org.onosproject.net.device.DeviceEvent;
 import org.onosproject.net.device.DeviceListener;
 import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.flow.DefaultTrafficSelector;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
 import org.onosproject.net.flow.TrafficSelector;
+import org.onosproject.net.flow.TrafficTreatment;
 import org.onosproject.net.host.DefaultHostDescription;
 import org.onosproject.net.host.HostDescription;
 import org.onosproject.net.host.HostProvider;
 import org.onosproject.net.host.HostProviderRegistry;
 import org.onosproject.net.host.HostProviderService;
 import org.onosproject.net.host.HostService;
+import org.onosproject.net.packet.DefaultOutboundPacket;
+import org.onosproject.net.packet.OutboundPacket;
 import org.onosproject.net.packet.PacketContext;
 import org.onosproject.net.packet.PacketPriority;
 import org.onosproject.net.packet.PacketProcessor;
@@ -66,6 +71,7 @@ import org.onosproject.net.topology.TopologyService;
 import org.osgi.service.component.ComponentContext;
 import org.slf4j.Logger;
 
+import java.nio.ByteBuffer;
 import java.util.Dictionary;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -126,6 +132,8 @@ public class HostLocationProvider extends AbstractProvider implements HostProvid
 
     protected ExecutorService eventHandler;
 
+    private static final byte[] SENDER_ADDRESS = IpAddress.valueOf("0.0.0.0").toOctets();
+
     /**
      * Creates an OpenFlow host provider.
      */
@@ -261,7 +269,60 @@ public class HostLocationProvider extends AbstractProvider implements HostProvid
 
     @Override
     public void triggerProbe(Host host) {
-        log.info("Triggering probe on device {}", host);
+        log.info("Triggering probe on device {} ", host);
+        MastershipRole role = deviceService.getRole(host.location().deviceId());
+        if (role.equals(MastershipRole.MASTER)) {
+            host.ipAddresses().forEach(ip -> {
+                sendProbe(host, ip);
+            });
+        } else {
+            log.info("not the master, master will probe {}");
+        }
+    }
+
+    private void sendProbe(Host host, IpAddress targetIp) {
+        Ethernet probePacket = null;
+        if (targetIp.isIp4()) {
+            // IPv4: Use ARP
+            probePacket = buildArpRequest(targetIp, host);
+        } else {
+            // IPv6: Use Neighbor Discovery
+            //FIX ME need to implement ndp probe
+            log.info("Triggering probe on device {} ", host);
+        }
+
+        TrafficTreatment treatment = DefaultTrafficTreatment.builder().setOutput(host.location().port()).build();
+
+        OutboundPacket outboundPacket = new DefaultOutboundPacket(host.location().deviceId(), treatment,
+                ByteBuffer.wrap(probePacket.serialize()));
+
+        packetService.emit(outboundPacket);
+    }
+
+    /*
+     * This method is using source ip as 0.0.0.0 , to receive the reply even from the sub net hosts.
+     */
+    private Ethernet buildArpRequest(IpAddress targetIp, Host host) {
+
+        ARP arp = new ARP();
+        arp.setHardwareType(ARP.HW_TYPE_ETHERNET)
+           .setHardwareAddressLength((byte) Ethernet.DATALAYER_ADDRESS_LENGTH)
+           .setProtocolType(ARP.PROTO_TYPE_IP)
+           .setProtocolAddressLength((byte) IpAddress.INET_BYTE_LENGTH)
+           .setOpCode(ARP.OP_REQUEST);
+
+        arp.setSenderHardwareAddress(MacAddress.BROADCAST.toBytes())
+                .setSenderProtocolAddress(SENDER_ADDRESS)
+                .setTargetHardwareAddress(MacAddress.BROADCAST.toBytes())
+                .setTargetProtocolAddress(targetIp.toOctets());
+
+        Ethernet ethernet = new Ethernet();
+        ethernet.setEtherType(Ethernet.TYPE_ARP)
+                .setDestinationMACAddress(MacAddress.BROADCAST)
+                .setSourceMACAddress(MacAddress.BROADCAST).setPayload(arp);
+
+        ethernet.setPad(true);
+        return ethernet;
     }
 
     private class InternalHostProvider implements PacketProcessor {

@@ -361,6 +361,26 @@ public abstract class Tools {
     }
 
     /**
+     * Get Long property from the propertyName
+     * Return null if propertyName is not found.
+     *
+     * @param properties   properties to be looked up
+     * @param propertyName the name of the property to look up
+     * @return value when the propertyName is defined or return null
+     */
+    public static Long getLongProperty(Dictionary<?, ?> properties,
+                                             String propertyName) {
+        Long value;
+        try {
+            String s = get(properties, propertyName);
+            value = Strings.isNullOrEmpty(s) ? null : Long.valueOf(s);
+        } catch (NumberFormatException | ClassCastException e) {
+            value = null;
+        }
+        return value;
+    }
+
+    /**
      * Returns a function that retries execution on failure.
      * @param base base function
      * @param exceptionClass type of exception for which to retry