opendaylight/arphandler/src/main/java/org/opendaylight/controller/arphandler/internal/ArpHandler.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 /*
  10  *
  11  */
  12 package org.opendaylight.controller.arphandler.internal;
  13
  14 import java.net.InetAddress;
  15 import java.net.UnknownHostException;
  16 import java.util.Arrays;
  17 import java.util.Collections;
  18 import java.util.EnumSet;
  19 import java.util.HashSet;
  20 import java.util.Set;
  21 import java.util.Timer;
  22 import java.util.TimerTask;
  23 import java.util.concurrent.BlockingQueue;
  24 import java.util.concurrent.ConcurrentHashMap;
  25 import java.util.concurrent.ConcurrentMap;
  26 import java.util.concurrent.CopyOnWriteArraySet;
  27 import java.util.concurrent.LinkedBlockingQueue;
  28
  29 import org.opendaylight.controller.arphandler.ARPCacheEvent;
  30 import org.opendaylight.controller.arphandler.ARPEvent;
  31 import org.opendaylight.controller.arphandler.ARPReply;
  32 import org.opendaylight.controller.arphandler.ARPRequest;
  33 import org.opendaylight.controller.clustering.services.CacheConfigException;
  34 import org.opendaylight.controller.clustering.services.CacheExistException;
  35 import org.opendaylight.controller.clustering.services.ICacheUpdateAware;
  36 import org.opendaylight.controller.clustering.services.IClusterContainerServices;
  37 import org.opendaylight.controller.clustering.services.IClusterServices;
  38 import org.opendaylight.controller.connectionmanager.IConnectionManager;
  39 import org.opendaylight.controller.hosttracker.HostIdFactory;
  40 import org.opendaylight.controller.hosttracker.IHostId;
  41 import org.opendaylight.controller.hosttracker.IfHostListener;
  42 import org.opendaylight.controller.hosttracker.IfIptoHost;
  43 import org.opendaylight.controller.hosttracker.hostAware.HostNodeConnector;
  44 import org.opendaylight.controller.hosttracker.hostAware.IHostFinder;
  45 import org.opendaylight.controller.sal.connection.ConnectionLocality;
  46 import org.opendaylight.controller.sal.core.ConstructionException;
  47 import org.opendaylight.controller.sal.core.Node;
  48 import org.opendaylight.controller.sal.core.NodeConnector;
  49 import org.opendaylight.controller.sal.packet.ARP;
  50 import org.opendaylight.controller.sal.packet.Ethernet;
  51 import org.opendaylight.controller.sal.packet.IDataPacketService;
  52 import org.opendaylight.controller.sal.packet.IListenDataPacket;
  53 import org.opendaylight.controller.sal.packet.IPv4;
  54 import org.opendaylight.controller.sal.packet.Packet;
  55 import org.opendaylight.controller.sal.packet.PacketResult;
  56 import org.opendaylight.controller.sal.packet.RawPacket;
  57 import org.opendaylight.controller.sal.routing.IRouting;
  58 import org.opendaylight.controller.sal.utils.EtherTypes;
  59 import org.opendaylight.controller.sal.utils.HexEncode;
  60 import org.opendaylight.controller.sal.utils.NetUtils;
  61 import org.opendaylight.controller.switchmanager.ISwitchManager;
  62 import org.opendaylight.controller.switchmanager.Subnet;
  63 import org.opendaylight.controller.topologymanager.ITopologyManager;
  64 import org.slf4j.Logger;
  65 import org.slf4j.LoggerFactory;
  66
  67 public class ArpHandler implements IHostFinder, IListenDataPacket, ICacheUpdateAware<ARPEvent, Boolean> {
  68     private static final Logger log = LoggerFactory.getLogger(ArpHandler.class);
  69     static final String ARP_EVENT_CACHE_NAME = "arphandler.arpRequestReplyEvent";
  70     private IfIptoHost hostTracker;
  71     private ISwitchManager switchManager;
  72     private ITopologyManager topologyManager;
  73     private IDataPacketService dataPacketService;
  74     private IRouting routing;
  75     private IClusterContainerServices clusterContainerService;
  76     private IConnectionManager connectionManager;
  77     private Set<IfHostListener> hostListeners = new CopyOnWriteArraySet<IfHostListener>();
  78     private ConcurrentMap<InetAddress, Set<HostNodeConnector>> arpRequestors;
  79     private ConcurrentMap<InetAddress, Short> countDownTimers;
  80     private Timer periodicTimer;
  81     private BlockingQueue<ARPCacheEvent> ARPCacheEvents = new LinkedBlockingQueue<ARPCacheEvent>();
  82     private Thread cacheEventHandler;
  83     private boolean stopping = false;
  84
  85     /*
  86      * A cluster allocated cache. Used for synchronizing ARP request/reply
  87      * events across all cluster controllers. To raise an event, we put() a
  88      * specific event object (as key) and all nodes handle it in the
  89      * entryUpdated callback.
  90      *
  91      * In case of ARPReply, we put true value to send replies to any requestors
  92      * by calling generateAndSendReply
  93      */
  94     private ConcurrentMap<ARPEvent, Boolean> arpRequestReplyEvent;
  95
  96     void setConnectionManager(IConnectionManager cm) {
  97         this.connectionManager = cm;
  98     }
  99
 100     void unsetConnectionManager(IConnectionManager cm) {
 101         if (this.connectionManager == cm) {
 102             connectionManager = null;
 103         }
 104     }
 105
 106     void setClusterContainerService(IClusterContainerServices s) {
 107         this.clusterContainerService = s;
 108     }
 109
 110     void unsetClusterContainerService(IClusterContainerServices s) {
 111         if (this.clusterContainerService == s) {
 112             this.clusterContainerService = null;
 113         }
 114     }
 115
 116     void setHostListener(IfHostListener s) {
 117         if (this.hostListeners != null) {
 118             this.hostListeners.add(s);
 119         }
 120     }
 121
 122     void unsetHostListener(IfHostListener s) {
 123         if (this.hostListeners != null) {
 124             this.hostListeners.remove(s);
 125         }
 126     }
 127
 128     void setDataPacketService(IDataPacketService s) {
 129         this.dataPacketService = s;
 130     }
 131
 132     void unsetDataPacketService(IDataPacketService s) {
 133         if (this.dataPacketService == s) {
 134             this.dataPacketService = null;
 135         }
 136     }
 137
 138     public void setHostTracker(IfIptoHost hostTracker) {
 139         log.debug("Setting HostTracker");
 140         this.hostTracker = hostTracker;
 141     }
 142
 143     public void unsetHostTracker(IfIptoHost s) {
 144         log.debug("UNSetting HostTracker");
 145         if (this.hostTracker == s) {
 146             this.hostTracker = null;
 147         }
 148     }
 149
 150     public void setTopologyManager(ITopologyManager tm) {
 151         this.topologyManager = tm;
 152     }
 153
 154     public void unsetTopologyManager(ITopologyManager tm) {
 155         if (this.topologyManager == tm) {
 156             this.topologyManager = null;
 157         }
 158     }
 159
 160     protected void sendARPReply(NodeConnector p, byte[] sMAC, InetAddress sIP, byte[] tMAC, InetAddress tIP) {
 161         byte[] senderIP = sIP.getAddress();
 162         byte[] targetIP = tIP.getAddress();
 163         ARP arp = createARP(ARP.REPLY, sMAC, senderIP, tMAC, targetIP);
 164
 165         if(log.isTraceEnabled()) {
 166             log.trace("Sending Arp Reply with srcMac {} - srcIp {} - dstMac {} - dstIp {} - outport {}",
 167                     HexEncode.bytesToHexString(sMAC),
 168                     sIP, HexEncode.bytesToHexString(tMAC), tIP, p);
 169         }
 170
 171         Ethernet ethernet = createEthernet(sMAC, tMAC, arp);
 172
 173         RawPacket destPkt = this.dataPacketService.encodeDataPacket(ethernet);
 174         destPkt.setOutgoingNodeConnector(p);
 175
 176         this.dataPacketService.transmitDataPacket(destPkt);
 177     }
 178
 179     private void logArpPacket(ARP pkt, NodeConnector p) {
 180         try {
 181             if (pkt.getOpCode() == ARP.REQUEST) {
 182                 log.trace("Received Arp Request with srcMac {} - srcIp {} - dstMac {} - dstIp {} - inport {}", HexEncode.bytesToHexString(pkt.getSenderHardwareAddress()),
 183                         InetAddress.getByAddress(pkt.getSenderProtocolAddress()), HexEncode.bytesToHexString(pkt.getTargetHardwareAddress()),
 184                         InetAddress.getByAddress(pkt.getTargetProtocolAddress()), p);
 185             } else if(pkt.getOpCode() == ARP.REPLY) {
 186                 log.trace("Received Arp Reply with srcMac {} - srcIp {} - dstMac {} - dstIp {} - inport {}", HexEncode.bytesToHexString(pkt.getSenderHardwareAddress()),
 187                         InetAddress.getByAddress(pkt.getSenderProtocolAddress()), HexEncode.bytesToHexString(pkt.getTargetHardwareAddress()),
 188                         InetAddress.getByAddress(pkt.getTargetProtocolAddress()), p);
 189             }
 190         } catch(UnknownHostException e) {
 191             log.warn("Illegal Ip Address in the ARP packet", e);
 192         }
 193     }
 194
 195     protected void handleARPPacket(Ethernet eHeader, ARP pkt, NodeConnector p) {
 196
 197         if(log.isTraceEnabled()) {
 198             logArpPacket(pkt, p);
 199         }
 200
 201         byte[] sourceMAC = eHeader.getSourceMACAddress();
 202         byte[] targetMAC = eHeader.getDestinationMACAddress();
 203         /*
 204          * Sanity Check; drop ARP packets originated by the controller itself.
 205          * This is to avoid continuous flooding
 206          */
 207         if (Arrays.equals(sourceMAC, getControllerMAC())) {
 208             if (log.isDebugEnabled()) {
 209                 log.debug("Receive a self originated ARP pkt (srcMAC {}) --> DROP",
 210                         HexEncode.bytesToHexString(sourceMAC));
 211             }
 212             return;
 213         }
 214
 215         InetAddress targetIP, sourceIP;
 216         try {
 217             targetIP = InetAddress.getByAddress(pkt.getTargetProtocolAddress());
 218             sourceIP = InetAddress.getByAddress(pkt.getSenderProtocolAddress());
 219         } catch (UnknownHostException e1) {
 220             log.debug("Invalid host in ARP packet: {}", e1.getMessage());
 221             return;
 222         }
 223
 224         Subnet subnet = null;
 225         if (switchManager != null) {
 226             subnet = switchManager.getSubnetByNetworkAddress(sourceIP);
 227         }
 228         if (subnet == null) {
 229             log.debug("ARPHandler: can't find subnet matching {}, drop packet", sourceIP);
 230             return;
 231         }
 232
 233         // Make sure that the host is a legitimate member of this subnet
 234         if (!subnet.hasNodeConnector(p)) {
 235             log.debug("{} showing up on {} does not belong to {}", new Object[] { sourceIP, p, subnet });
 236             return;
 237         }
 238
 239         HostNodeConnector requestor = null;
 240         if (NetUtils.isUnicastMACAddr(sourceMAC) && p.getNode() != null) {
 241             try {
 242                 requestor = new HostNodeConnector(sourceMAC, sourceIP, p, subnet.getVlan());
 243             } catch (ConstructionException e) {
 244                 log.debug("Received ARP packet with invalid MAC: {}", HexEncode.bytesToHexString(sourceMAC));
 245                 return;
 246             }
 247             /*
 248              * Learn host from the received ARP REQ/REPLY, inform Host Tracker
 249              */
 250             log.trace("Inform Host tracker of new host {}", requestor.getNetworkAddress());
 251             for (IfHostListener listener : this.hostListeners) {
 252                 listener.hostListener(requestor);
 253             }
 254         }
 255
 256         /*
 257          * OpCode != request -> ARP Reply. If there are hosts (in arpRequestors)
 258          * waiting for the ARP reply for this sourceIP, it's time to generate
 259          * the reply and send it to these hosts.
 260          *
 261          * If sourceIP==targetIP, it is a Gratuitous ARP. If there are hosts (in
 262          * arpRequestors) waiting for the ARP reply for this sourceIP, it's time
 263          * to generate the reply and send it to these hosts
 264          */
 265
 266         if (pkt.getOpCode() != ARP.REQUEST || sourceIP.equals(targetIP)) {
 267             // Raise a reply event so that any waiting requestors will be sent a
 268             // reply
 269             // the true value indicates we should generate replies to requestors
 270             // across the cluster
 271             log.trace("Received ARP reply packet from {}, reply to all requestors.", sourceIP);
 272             arpRequestReplyEvent.put(new ARPReply(sourceIP, sourceMAC), true);
 273             return;
 274         }
 275
 276         /*
 277          * ARP Request Handling: If targetIP is the IP of the subnet, reply with
 278          * ARP REPLY If targetIP is a known host, PROXY ARP (by sending ARP
 279          * REPLY) on behalf of known target hosts. For unknown target hosts,
 280          * generate and send an ARP request to ALL switches/ports using the IP
 281          * address defined in the subnet as source address
 282          */
 283         /*
 284          * If target IP is gateway IP, Send ARP reply
 285          */
 286         if ((targetIP.equals(subnet.getNetworkAddress()))
 287                 && (NetUtils.isBroadcastMACAddr(targetMAC) || Arrays.equals(targetMAC, getControllerMAC()))) {
 288             if (connectionManager.getLocalityStatus(p.getNode()) == ConnectionLocality.LOCAL) {
 289                 if (log.isTraceEnabled()) {
 290                     log.trace("Received local ARP req. for default gateway. Replying with controller MAC: {}",
 291                             HexEncode.bytesToHexString(getControllerMAC()));
 292                 }
 293                 sendARPReply(p, getControllerMAC(), targetIP, pkt.getSenderHardwareAddress(), sourceIP);
 294             } else {
 295                 log.trace("Received non-local ARP req. for default gateway. Raising reply event");
 296                 arpRequestReplyEvent.put(
 297                         new ARPReply(p, targetIP, getControllerMAC(), sourceIP, pkt.getSenderHardwareAddress()), false);
 298             }
 299             return;
 300         }
 301
 302         // Hosttracker hosts db key implementation
 303         IHostId id = HostIdFactory.create(targetIP, null);
 304         HostNodeConnector host = hostTracker.hostQuery(id);
 305         // unknown host, initiate ARP request
 306         if (host == null) {
 307             // add the requestor to the list so that we can replay the reply
 308             // when the host responds
 309             if (requestor != null) {
 310                 Set<HostNodeConnector> requestorSet = arpRequestors.get(targetIP);
 311                 if (requestorSet == null) {
 312                     requestorSet = Collections.newSetFromMap(new ConcurrentHashMap<HostNodeConnector, Boolean>());
 313                     arpRequestors.put(targetIP, requestorSet);
 314                 }
 315                 requestorSet.add(requestor);
 316                 countDownTimers.put(targetIP, (short) 2); // reset timeout to
 317                                                           // 2sec
 318             }
 319             // Raise a bcast request event, all controllers need to send one
 320             log.trace("Sending a bcast ARP request for {}", targetIP);
 321             arpRequestReplyEvent.put(new ARPRequest(targetIP, subnet), false);
 322         } else {
 323             /*
 324              * Target host known (across the cluster), send ARP REPLY make sure
 325              * that targetMAC matches the host's MAC if it is not broadcastMAC
 326              */
 327             if (NetUtils.isBroadcastMACAddr(targetMAC) || Arrays.equals(host.getDataLayerAddressBytes(), targetMAC)) {
 328                 log.trace("Received ARP req. for known host {}, sending reply...", targetIP);
 329                 if (connectionManager.getLocalityStatus(p.getNode()) == ConnectionLocality.LOCAL) {
 330                     sendARPReply(p, host.getDataLayerAddressBytes(), host.getNetworkAddress(),
 331                             pkt.getSenderHardwareAddress(), sourceIP);
 332                 } else {
 333                     arpRequestReplyEvent.put(new ARPReply(p, host.getNetworkAddress(), host.getDataLayerAddressBytes(),
 334                             sourceIP, pkt.getSenderHardwareAddress()), false);
 335                 }
 336             } else {
 337                 /*
 338                  * Target MAC has been changed. For now, discard it. TODO: We
 339                  * may need to send unicast ARP REQUEST on behalf of the target
 340                  * back to the sender to trigger the sender to update its table
 341                  */
 342             }
 343         }
 344     }
 345
 346     /**
 347      * Send a broadcast ARP Request to the switch/ ports using the
 348      * networkAddress of the subnet as sender IP the controller's MAC as sender
 349      * MAC the targetIP as the target Network Address
 350      */
 351     protected void sendBcastARPRequest(InetAddress targetIP, Subnet subnet) {
 352         log.trace("sendBcatARPRequest targetIP:{} subnet:{}", targetIP, subnet);
 353         Set<NodeConnector> nodeConnectors;
 354         if (subnet.isFlatLayer2()) {
 355             nodeConnectors = new HashSet<NodeConnector>();
 356             for (Node n : this.switchManager.getNodes()) {
 357                 nodeConnectors.addAll(this.switchManager.getUpNodeConnectors(n));
 358             }
 359         } else {
 360             nodeConnectors = subnet.getNodeConnectors();
 361         }
 362         byte[] targetHardwareAddress = new byte[] { (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0 };
 363
 364         // TODO: should use IBroadcastHandler instead
 365         for (NodeConnector p : nodeConnectors) {
 366             // filter out any non-local or internal ports
 367             if (!(connectionManager.getLocalityStatus(p.getNode()) == ConnectionLocality.LOCAL)
 368                     || topologyManager.isInternal(p)) {
 369                 continue;
 370             }
 371             log.trace("Sending toward nodeConnector:{}", p);
 372             byte[] senderIP = subnet.getNetworkAddress().getAddress();
 373             byte[] targetIPByte = targetIP.getAddress();
 374             ARP arp = createARP(ARP.REQUEST, getControllerMAC(), senderIP, targetHardwareAddress, targetIPByte);
 375
 376             if(log.isTraceEnabled()) {
 377                 log.trace("Sending Broadcast Arp Request with srcMac {} - srcIp {} - dstMac {} - dstIp {} - outport {}", HexEncode.bytesToHexString(getControllerMAC()),
 378                         subnet.getNetworkAddress(), HexEncode.bytesToHexString(targetHardwareAddress), targetIP, p);
 379             }
 380
 381             byte[] destMACAddress = NetUtils.getBroadcastMACAddr();
 382             Ethernet ethernet = createEthernet(getControllerMAC(), destMACAddress, arp);
 383
 384             // TODO For now send port-by-port, see how to optimize to
 385             // send to multiple ports at once
 386             RawPacket destPkt = this.dataPacketService.encodeDataPacket(ethernet);
 387             destPkt.setOutgoingNodeConnector(p);
 388
 389             this.dataPacketService.transmitDataPacket(destPkt);
 390         }
 391     }
 392
 393     /**
 394      * Send a unicast ARP Request to the known host on a specific switch/port as
 395      * defined in the host. The sender IP is the networkAddress of the subnet
 396      * The sender MAC is the controller's MAC
 397      */
 398     protected void sendUcastARPRequest(HostNodeConnector host, Subnet subnet) {
 399         log.trace("sendUcastARPRequest host:{} subnet:{}", host, subnet);
 400         NodeConnector outPort = host.getnodeConnector();
 401         if (outPort == null) {
 402             log.error("Failed sending UcastARP because cannot extract output port from Host: {}", host);
 403             return;
 404         }
 405
 406         byte[] senderIP = subnet.getNetworkAddress().getAddress();
 407         byte[] targetIP = host.getNetworkAddress().getAddress();
 408         byte[] targetMAC = host.getDataLayerAddressBytes();
 409         ARP arp = createARP(ARP.REQUEST, getControllerMAC(), senderIP, targetMAC, targetIP);
 410
 411         if(log.isTraceEnabled()) {
 412             log.trace("Sending Unicast Arp Request with srcMac {} - srcIp {} - dstMac {} - dstIp {} - outport {}",
 413                     HexEncode.bytesToHexString(getControllerMAC()),
 414                     subnet.getNetworkAddress(), HexEncode.bytesToHexString(targetMAC), host.getNetworkAddress(),
 415                     outPort);
 416         }
 417
 418         Ethernet ethernet = createEthernet(getControllerMAC(), targetMAC, arp);
 419
 420         RawPacket destPkt = this.dataPacketService.encodeDataPacket(ethernet);
 421         destPkt.setOutgoingNodeConnector(outPort);
 422
 423         this.dataPacketService.transmitDataPacket(destPkt);
 424     }
 425
 426     @Override
 427     public void find(InetAddress networkAddress) {
 428         log.trace("Received find IP {}", networkAddress);
 429
 430         Subnet subnet = null;
 431         if (switchManager != null) {
 432             subnet = switchManager.getSubnetByNetworkAddress(networkAddress);
 433         }
 434         if (subnet == null) {
 435             log.debug("Can't find subnet matching IP {}", networkAddress);
 436             return;
 437         }
 438
 439         // send a broadcast ARP Request to this IP
 440         arpRequestReplyEvent.put(new ARPRequest(networkAddress, subnet), false);
 441     }
 442
 443     /*
 444      * Probe the host by sending a unicast ARP Request to the host
 445      */
 446     @Override
 447     public void probe(HostNodeConnector host) {
 448         log.trace("Received probe host {}", host);
 449
 450         Subnet subnet = null;
 451         if (switchManager != null) {
 452             subnet = switchManager.getSubnetByNetworkAddress(host.getNetworkAddress());
 453         }
 454         if (subnet == null) {
 455             log.debug("can't find subnet matching {}", host.getNetworkAddress());
 456             return;
 457         }
 458
 459         if (connectionManager.getLocalityStatus(host.getnodeconnectorNode()) == ConnectionLocality.LOCAL) {
 460             log.trace("Send a ucast ARP req. to: {}", host);
 461             sendUcastARPRequest(host, subnet);
 462         } else {
 463             log.trace("Raise a ucast ARP req. event to: {}", host);
 464             arpRequestReplyEvent.put(new ARPRequest(host, subnet), false);
 465         }
 466     }
 467
 468     /**
 469      * An IP packet is punted to the controller, this means that the destination
 470      * host is not known to the controller. Need to discover it by sending a
 471      * Broadcast ARP Request
 472      *
 473      * @param pkt
 474      * @param p
 475      */
 476     protected void handlePuntedIPPacket(IPv4 pkt, NodeConnector p) {
 477
 478         InetAddress dIP = NetUtils.getInetAddress(pkt.getDestinationAddress());
 479         if (dIP == null) {
 480             return;
 481         }
 482
 483         // try to find a matching subnet
 484         Subnet subnet = null;
 485         if (switchManager != null) {
 486             subnet = switchManager.getSubnetByNetworkAddress(dIP);
 487         }
 488         if (subnet == null) {
 489             log.debug("Can't find subnet matching {}, drop packet", dIP);
 490             return;
 491         }
 492         // If packet is sent to the default gw (us), ignore it for now
 493         if (subnet.getNetworkAddress().equals(dIP)) {
 494             log.trace("Ignore IP packet destined to default gw");
 495             return;
 496         }
 497
 498         // see if we know about the host
 499         // Hosttracker hosts db key implementation
 500         HostNodeConnector host = hostTracker.hostFind(dIP);
 501
 502         if (host == null) {
 503             // if we don't know about the host, try to find it
 504             log.trace("Punted IP pkt to {}, sending bcast ARP event...", dIP);
 505             /*
 506              * unknown destination host, initiate bcast ARP request
 507              */
 508             arpRequestReplyEvent.put(new ARPRequest(dIP, subnet), false);
 509
 510         } else {
 511             log.trace("Ignoring punted IP pkt to known host: {} (received on: {})", dIP, p);
 512         }
 513     }
 514
 515     public byte[] getControllerMAC() {
 516         if (switchManager == null) {
 517             return null;
 518         }
 519         return switchManager.getControllerMAC();
 520     }
 521
 522     /**
 523      * Function called by the dependency manager when all the required
 524      * dependencies are satisfied
 525      *
 526      */
 527     void init() {
 528         arpRequestors = new ConcurrentHashMap<InetAddress, Set<HostNodeConnector>>();
 529         countDownTimers = new ConcurrentHashMap<InetAddress, Short>();
 530         cacheEventHandler = new Thread(new ARPCacheEventHandler(), "ARPCacheEventHandler Thread");
 531
 532         allocateCaches();
 533         retrieveCaches();
 534
 535     }
 536
 537     @SuppressWarnings({ "unchecked" })
 538     private void retrieveCaches() {
 539         ConcurrentMap<?, ?> map;
 540
 541         if (this.clusterContainerService == null) {
 542             log.error("Cluster service unavailable, can't retieve ARPHandler caches!");
 543             return;
 544         }
 545
 546         map = clusterContainerService.getCache(ARP_EVENT_CACHE_NAME);
 547         if (map != null) {
 548             this.arpRequestReplyEvent = (ConcurrentMap<ARPEvent, Boolean>) map;
 549         } else {
 550             log.error("Cache allocation failed for {}", ARP_EVENT_CACHE_NAME);
 551         }
 552     }
 553
 554     private void allocateCaches() {
 555         if (clusterContainerService == null) {
 556             nonClusterObjectCreate();
 557             log.error("Clustering service unavailable. Allocated non-cluster caches for ARPHandler.");
 558             return;
 559         }
 560
 561         try {
 562             clusterContainerService.createCache(ARP_EVENT_CACHE_NAME,
 563                     EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 564         } catch (CacheConfigException e) {
 565             log.error("ARPHandler cache configuration invalid!");
 566         } catch (CacheExistException e) {
 567             log.debug("ARPHandler cache exists, skipped allocation.");
 568         }
 569
 570     }
 571
 572     private void nonClusterObjectCreate() {
 573         arpRequestReplyEvent = new ConcurrentHashMap<ARPEvent, Boolean>();
 574     }
 575
 576     /**
 577      * Function called by the dependency manager when at least one dependency
 578      * become unsatisfied or when the component is shutting down because for
 579      * example bundle is being stopped.
 580      *
 581      */
 582     void destroy() {
 583         cacheEventHandler.interrupt();
 584     }
 585
 586     /**
 587      * Function called by dependency manager after "init ()" is called and after
 588      * the services provided by the class are registered in the service registry
 589      *
 590      */
 591     void start() {
 592         stopping = false;
 593         startPeriodicTimer();
 594         cacheEventHandler.start();
 595     }
 596
 597     /**
 598      * Function called by the dependency manager before the services exported by
 599      * the component are unregistered, this will be followed by a "destroy ()"
 600      * calls
 601      *
 602      */
 603     void stop() {
 604     }
 605
 606     void stopping() {
 607         stopping = true;
 608         cancelPeriodicTimer();
 609     }
 610
 611     void setSwitchManager(ISwitchManager s) {
 612         log.debug("SwitchManager service set.");
 613         this.switchManager = s;
 614     }
 615
 616     void unsetSwitchManager(ISwitchManager s) {
 617         if (this.switchManager == s) {
 618             log.debug("SwitchManager service UNset.");
 619             this.switchManager = null;
 620         }
 621     }
 622
 623     @Override
 624     public PacketResult receiveDataPacket(RawPacket inPkt) {
 625         if (inPkt == null) {
 626             return PacketResult.IGNORED;
 627         }
 628         log.trace("Received a frame of size: {}", inPkt.getPacketData().length);
 629         Packet formattedPak = this.dataPacketService.decodeDataPacket(inPkt);
 630         if (formattedPak instanceof Ethernet) {
 631             Object nextPak = formattedPak.getPayload();
 632             if (nextPak instanceof IPv4) {
 633                 log.trace("Handle IP packet: {}", formattedPak);
 634                 handlePuntedIPPacket((IPv4) nextPak, inPkt.getIncomingNodeConnector());
 635             } else if (nextPak instanceof ARP) {
 636                 log.trace("Handle ARP packet: {}", formattedPak);
 637                 handleARPPacket((Ethernet) formattedPak, (ARP) nextPak, inPkt.getIncomingNodeConnector());
 638             }
 639         }
 640         return PacketResult.IGNORED;
 641     }
 642
 643     private ARP createARP(short opCode, byte[] senderMacAddress, byte[] senderIP, byte[] targetMacAddress,
 644             byte[] targetIP) {
 645         ARP arp = new ARP();
 646         arp.setHardwareType(ARP.HW_TYPE_ETHERNET);
 647         arp.setProtocolType(EtherTypes.IPv4.shortValue());
 648         arp.setHardwareAddressLength((byte) 6);
 649         arp.setProtocolAddressLength((byte) 4);
 650         arp.setOpCode(opCode);
 651         arp.setSenderHardwareAddress(senderMacAddress);
 652         arp.setSenderProtocolAddress(senderIP);
 653         arp.setTargetHardwareAddress(targetMacAddress);
 654         arp.setTargetProtocolAddress(targetIP);
 655         return arp;
 656     }
 657
 658     private Ethernet createEthernet(byte[] sourceMAC, byte[] targetMAC, ARP arp) {
 659         Ethernet ethernet = new Ethernet();
 660         ethernet.setSourceMACAddress(sourceMAC);
 661         ethernet.setDestinationMACAddress(targetMAC);
 662         ethernet.setEtherType(EtherTypes.ARP.shortValue());
 663         ethernet.setPayload(arp);
 664         return ethernet;
 665     }
 666
 667     private void startPeriodicTimer() {
 668         this.periodicTimer = new Timer("ArpHandler Periodic Timer");
 669         this.periodicTimer.scheduleAtFixedRate(new TimerTask() {
 670             @Override
 671             public void run() {
 672                 Set<InetAddress> targetIPs = countDownTimers.keySet();
 673                 Set<InetAddress> expiredTargets = new HashSet<InetAddress>();
 674                 for (InetAddress t : targetIPs) {
 675                     short tick = countDownTimers.get(t);
 676                     tick--;
 677                     if (tick <= 0) {
 678                         expiredTargets.add(t);
 679                     } else {
 680                         countDownTimers.replace(t, tick);
 681                     }
 682                 }
 683                 for (InetAddress tIP : expiredTargets) {
 684                     countDownTimers.remove(tIP);
 685                     // Remove the requestor(s) who have been waiting for the ARP
 686                     // reply from this target for more than 1sec
 687                     arpRequestors.remove(tIP);
 688                     log.debug("ARP reply was not received from {}", tIP);
 689                 }
 690
 691                 // Clean up ARP event cache
 692                 try {
 693                     if (clusterContainerService.amICoordinator() && !arpRequestReplyEvent.isEmpty()) {
 694                         arpRequestReplyEvent.clear();
 695                     }
 696                 } catch (Exception e) {
 697                     log.warn("ARPHandler: A cluster member failed to clear event cache.");
 698                 }
 699             }
 700         }, 0, 1000);
 701     }
 702
 703     private void cancelPeriodicTimer() {
 704         if (this.periodicTimer != null) {
 705             this.periodicTimer.cancel();
 706         }
 707     }
 708
 709     private void generateAndSendReply(InetAddress sourceIP, byte[] sourceMAC) {
 710         if (log.isTraceEnabled()) {
 711             log.trace("generateAndSendReply called with params sourceIP:{} sourceMAC:{}", sourceIP,
 712                     HexEncode.bytesToHexString(sourceMAC));
 713         }
 714         Set<HostNodeConnector> hosts = arpRequestors.remove(sourceIP);
 715         if ((hosts == null) || hosts.isEmpty()) {
 716             log.trace("Bailing out no requestors Hosts");
 717             return;
 718         }
 719         countDownTimers.remove(sourceIP);
 720         for (HostNodeConnector host : hosts) {
 721             if (log.isTraceEnabled()) {
 722                 log.trace(
 723                         "Sending ARP Reply with src {}/{}, target {}/{}",
 724                         new Object[] { HexEncode.bytesToHexString(sourceMAC), sourceIP,
 725                                 HexEncode.bytesToHexString(host.getDataLayerAddressBytes()), host.getNetworkAddress() });
 726             }
 727             if (connectionManager.getLocalityStatus(host.getnodeconnectorNode()) == ConnectionLocality.LOCAL) {
 728                 sendARPReply(host.getnodeConnector(), sourceMAC, sourceIP, host.getDataLayerAddressBytes(),
 729                         host.getNetworkAddress());
 730             } else {
 731                 /*
 732                  * In the remote event a requestor moved to another controller
 733                  * it may turn out it now we need to send the ARP reply from a
 734                  * different controller, this cover the case
 735                  */
 736                 arpRequestReplyEvent.put(
 737                         new ARPReply(host.getnodeConnector(), sourceIP, sourceMAC, host.getNetworkAddress(), host
 738                                 .getDataLayerAddressBytes()), false);
 739             }
 740         }
 741     }
 742
 743     @Override
 744     public void entryUpdated(ARPEvent key, Boolean new_value, String cacheName, boolean originLocal) {
 745         log.trace("Got and entryUpdated for cacheName {} key {} isNew {}", cacheName, key, new_value);
 746         enqueueARPCacheEvent(key, new_value);
 747     }
 748
 749     @Override
 750     public void entryCreated(ARPEvent key, String cacheName, boolean originLocal) {
 751         // nothing to do
 752     }
 753
 754     @Override
 755     public void entryDeleted(ARPEvent key, String cacheName, boolean originLocal) {
 756         // nothing to do
 757     }
 758
 759     private void enqueueARPCacheEvent(ARPEvent event, boolean new_value) {
 760         try {
 761             ARPCacheEvent cacheEvent = new ARPCacheEvent(event, new_value);
 762             if (!ARPCacheEvents.contains(cacheEvent)) {
 763                 this.ARPCacheEvents.add(cacheEvent);
 764                 log.trace("Enqueued {}", event);
 765             }
 766         } catch (Exception e) {
 767             log.debug("enqueueARPCacheEvent caught Interrupt Exception for event {}", event);
 768         }
 769     }
 770
 771     /*
 772      * this thread monitors the connectionEvent queue for new incoming events
 773      * from
 774      */
 775     private class ARPCacheEventHandler implements Runnable {
 776         @Override
 777         public void run() {
 778             while (!stopping) {
 779                 try {
 780                     ARPCacheEvent ev = ARPCacheEvents.take();
 781                     ARPEvent event = ev.getEvent();
 782                     if (event instanceof ARPRequest) {
 783                         ARPRequest req = (ARPRequest) event;
 784                         // If broadcast request
 785                         if (req.getHost() == null) {
 786                             log.trace("Trigger and ARP Broadcast Request upon receipt of {}", req);
 787                             sendBcastARPRequest(req.getTargetIP(), req.getSubnet());
 788
 789                             // If unicast and local, send reply
 790                         } else if (connectionManager.getLocalityStatus(req.getHost().getnodeconnectorNode()) == ConnectionLocality.LOCAL) {
 791                             log.trace("ARPCacheEventHandler - sendUcatARPRequest upon receipt of {}", req);
 792                             sendUcastARPRequest(req.getHost(), req.getSubnet());
 793                         }
 794                     } else if (event instanceof ARPReply) {
 795                         ARPReply rep = (ARPReply) event;
 796                         // New reply received by controller, notify all awaiting
 797                         // requestors across the cluster
 798                         if (ev.isNewReply()) {
 799                             log.trace("Trigger a generateAndSendReply in response to {}", rep);
 800                             generateAndSendReply(rep.getTargetIP(), rep.getTargetMac());
 801                             // Otherwise, a specific reply. If local, send out.
 802                         } else if (connectionManager.getLocalityStatus(rep.getPort().getNode()) == ConnectionLocality.LOCAL) {
 803                             log.trace("ARPCacheEventHandler - sendUcatARPReply locally in response to {}", rep);
 804                             sendARPReply(rep.getPort(), rep.getSourceMac(), rep.getSourceIP(), rep.getTargetMac(),
 805                                     rep.getTargetIP());
 806                         }
 807                     }
 808                 } catch (InterruptedException e) {
 809                     ARPCacheEvents.clear();
 810                     return;
 811                 }
 812             }
 813         }
 814     }
 815 }