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